diff --git a/docs/operations/deep-storage-migration.md b/docs/operations/deep-storage-migration.md index 733db8bd924b..e72f44f5896f 100644 --- a/docs/operations/deep-storage-migration.md +++ b/docs/operations/deep-storage-migration.md @@ -39,6 +39,7 @@ To ensure a clean migration, shut down the non-coordinator services to ensure th change as you do the migration. When migrating from Derby, the coordinator processes will still need to be up initially, as they host the Derby database. +When migrating from PostgreSQL or another external metadata store, no Druid processes need to be running. ## Copy segments from old deep storage to new deep storage. @@ -48,7 +49,7 @@ For information on what path structure to use in the new deep storage, please se ## Export segments with rewritten load specs -Druid provides an [Export Metadata Tool](../operations/export-metadata.md) for exporting metadata from Derby into CSV files +Druid provides an [Export Metadata Tool](../operations/export-metadata.md) for exporting metadata from Derby or PostgreSQL into CSV files which can then be reimported. By setting [deep storage migration options](../operations/export-metadata.md#deep-storage-migration), the `export-metadata` tool will export CSV files where the segment load specs have been rewritten to load from your new deep storage location. diff --git a/docs/operations/export-metadata.md b/docs/operations/export-metadata.md index e065e42b0132..a152ddde980f 100644 --- a/docs/operations/export-metadata.md +++ b/docs/operations/export-metadata.md @@ -36,9 +36,10 @@ This tool exports the contents of the following Druid metadata tables: Additionally, the tool can rewrite the local deep storage location descriptors in the rows of the segments table to point to new deep storage locations (S3, HDFS, and local rewrite paths are supported). +The tool supports exporting from both Derby and PostgreSQL metadata stores. + The tool has the following limitations: -- Only exporting from Derby metadata is currently supported - If rewriting load specs for deep storage migration, only migrating from local deep storage is currently supported. ## `export-metadata` Options @@ -47,7 +48,7 @@ The `export-metadata` tool provides the following options: ### Connection Properties -- `--connectURI`: The URI of the Derby database, e.g. `jdbc:derby://localhost:1527/var/druid/metadata.db;create=true` +- `--connectURI`: The URI of the metadata database, e.g. `jdbc:derby://localhost:1527/var/druid/metadata.db;create=true` for Derby or `jdbc:postgresql://localhost:5432/druid` for PostgreSQL - `--user`: Username - `--password`: Password - `--base`: corresponds to the value of `druid.metadata.storage.tables.base` in the configuration, `druid` by default. @@ -133,7 +134,9 @@ If the new path was `/migration/example`, the contents of `/migration/example/` ## Running the tool -To use the tool, you can run the following from the root of the Druid package: +To use the tool, you can run the following from the root of the Druid package. + +### Exporting from Derby ```bash cd ${DRUID_ROOT} @@ -141,7 +144,17 @@ mkdir -p /tmp/csv java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList=[] org.apache.druid.cli.Main tools export-metadata --connectURI "jdbc:derby://localhost:1527/var/druid/metadata.db;" -o /tmp/csv ``` -In the example command above: +### Exporting from PostgreSQL + +When exporting from PostgreSQL, you must load the `postgresql-metadata-storage` extension and set the storage type to `postgresql`: + +```bash +cd ${DRUID_ROOT} +mkdir -p /tmp/csv +java -classpath "lib/*" -Dlog4j.configurationFile=conf/druid/cluster/_common/log4j2.xml -Ddruid.extensions.directory="extensions" -Ddruid.extensions.loadList='["postgresql-metadata-storage"]' -Ddruid.metadata.storage.type=postgresql org.apache.druid.cli.Main tools export-metadata --connectURI "jdbc:postgresql://localhost:5432/druid" --user druid --password druid -o /tmp/csv +``` + +In the example commands above: - `lib` is the Druid lib directory - `extensions` is the Druid extensions directory @@ -151,7 +164,7 @@ In the example command above: After running the tool, the output directory will contain `_raw.csv` and `.csv` files. -The `_raw.csv` files are intermediate files used by the tool, containing the table data as exported by Derby without modification. +The `_raw.csv` files are intermediate files used by the tool, containing the table data as exported from the source database without deep-storage rewrites. BLOB columns are hex-encoded and booleans are written as `true`/`false` strings. The `.csv` files are used for import into another database such as MySQL and PostgreSQL and have any configured deep storage location rewrites applied. diff --git a/docs/operations/metadata-migration.md b/docs/operations/metadata-migration.md index ea3596784ad2..01461ac750da 100644 --- a/docs/operations/metadata-migration.md +++ b/docs/operations/metadata-migration.md @@ -24,7 +24,8 @@ title: "Metadata Migration" If you have been running an evaluation Druid cluster using the built-in Derby metadata storage and wish to migrate to a -more production-capable metadata store such as MySQL or PostgreSQL, this document describes the necessary steps. +more production-capable metadata store such as MySQL or PostgreSQL, or if you need to migrate metadata between +production stores (e.g., from PostgreSQL to MySQL), this document describes the necessary steps. ## Shut down cluster services @@ -35,7 +36,7 @@ When migrating from Derby, the coordinator processes will still need to be up in ## Exporting metadata -Druid provides an [Export Metadata Tool](../operations/export-metadata.md) for exporting metadata from Derby into CSV files +Druid provides an [Export Metadata Tool](../operations/export-metadata.md) for exporting metadata from Derby or PostgreSQL into CSV files which can then be imported into your new metadata store. The tool also provides options for rewriting the deep storage locations of segments; this is useful diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java index 74829cddbfbf..e651276062c6 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java @@ -25,6 +25,7 @@ import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import com.google.common.io.BaseEncoding; import org.apache.commons.codec.digest.DigestUtils; import org.apache.commons.dbcp2.BasicDataSource; import org.apache.commons.dbcp2.BasicDataSourceFactory; @@ -50,12 +51,18 @@ import javax.annotation.Nullable; import javax.validation.constraints.NotNull; +import java.io.FileOutputStream; +import java.io.OutputStreamWriter; +import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.DatabaseMetaData; import java.sql.ResultSet; +import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.sql.SQLRecoverableException; import java.sql.SQLTransientException; +import java.sql.Statement; +import java.sql.Types; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -1038,6 +1045,84 @@ public void createAuditTable() } } + @Override + public void exportTable( + final String tableName, + final String outputPath + ) + { + exportTableWithJdbc(tableName, outputPath); + } + + /** + * Exports a table to a CSV file using generic JDBC. + * Binary columns are hex-encoded and booleans are written as true/false strings. + * Subclasses may override {@link #exportTable} with a database-specific implementation + * while this method remains available for testing or fallback. + */ + protected void exportTableWithJdbc( + final String tableName, + final String outputPath + ) + { + // Use a transaction so that the connection has autoCommit=false. + // PostgreSQL JDBC requires autoCommit=false and a positive fetch size + // to use cursor-based streaming instead of buffering the entire ResultSet. + retryTransaction( + (TransactionCallback) (handle, status) -> { + final Connection conn = handle.getConnection(); + try (Statement stmt = conn.createStatement()) { + final int fetchSize = getStreamingFetchSize(); + if (fetchSize > 0) { + stmt.setFetchSize(fetchSize); + } + try (ResultSet rs = stmt.executeQuery(StringUtils.format("SELECT * FROM %s", tableName)); + FileOutputStream fos = new FileOutputStream(outputPath); + OutputStreamWriter writer = new OutputStreamWriter(fos, StandardCharsets.UTF_8)) { + final ResultSetMetaData meta = rs.getMetaData(); + final int columnCount = meta.getColumnCount(); + while (rs.next()) { + for (int i = 1; i <= columnCount; i++) { + if (i > 1) { + writer.write(','); + } + final int colType = meta.getColumnType(i); + if (colType == Types.BINARY || colType == Types.VARBINARY + || colType == Types.LONGVARBINARY || colType == Types.BLOB + || (colType == Types.OTHER && "bytea".equalsIgnoreCase(meta.getColumnTypeName(i)))) { + final byte[] bytes = rs.getBytes(i); + if (bytes != null) { + writer.write(BaseEncoding.base16().encode(bytes)); + } + } else if (colType == Types.BOOLEAN || colType == Types.BIT) { + final boolean val = rs.getBoolean(i); + if (!rs.wasNull()) { + writer.write(String.valueOf(val)); + } + } else { + final String val = rs.getString(i); + if (val != null) { + if (val.contains(",") || val.contains("\"") || val.contains("\n") || val.contains("\r")) { + writer.write('"'); + writer.write(StringUtils.replace(val, "\"", "\"\"")); + writer.write('"'); + } else { + writer.write(val); + } + } + } + } + writer.write('\n'); + } + } + } + return null; + }, + QUIET_RETRIES, + DEFAULT_MAX_TRIES + ); + } + @Override public void deleteAllRecords(final String tableName) { diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java index 3fef28a963a6..86ab26df2c58 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java @@ -24,6 +24,7 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import com.google.common.collect.Sets; +import com.google.common.io.BaseEncoding; import org.apache.commons.dbcp2.BasicDataSource; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig; @@ -37,6 +38,10 @@ import org.skife.jdbi.v2.exceptions.UnableToExecuteStatementException; import org.skife.jdbi.v2.exceptions.UnableToObtainConnectionException; +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.sql.SQLException; import java.sql.SQLRecoverableException; import java.sql.SQLTransientConnectionException; @@ -463,6 +468,219 @@ private void assertIndicesPresentOnTable(String tableName, Set expectedI ); } + @Test + public void testExportTable() throws IOException + { + final String tableName = "test_export"; + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (name VARCHAR(255) NOT NULL, payload BLOB NOT NULL, active BOOLEAN NOT NULL, PRIMARY KEY(name))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?)", tableName), + "key1", + StringUtils.toUtf8("{\"type\":\"test\"}"), + true + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?)", tableName), + "key2", + StringUtils.toUtf8("{\"value\":42}"), + false + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_test", ".csv").toFile(); + outputFile.deleteOnExit(); + + // Call the base class exportTable (the generic JDBC path used by PostgreSQL) + // rather than DerbyConnector's native SYSCS_EXPORT_TABLE override + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath() + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(2, lines.size()); + Collections.sort(lines); + + // Verify rows (sorted by name): hex-encoded payload, boolean as string + final String expectedHex1 = BaseEncoding.base16().encode(StringUtils.toUtf8("{\"type\":\"test\"}")); + Assert.assertEquals("key1," + expectedHex1 + ",true", lines.get(0)); + + final String expectedHex2 = BaseEncoding.base16().encode(StringUtils.toUtf8("{\"value\":42}")); + Assert.assertEquals("key2," + expectedHex2 + ",false", lines.get(1)); + + dropTable(tableName); + } + + @Test + public void testExportTableWithSpecialCharacters() throws IOException + { + final String tableName = "test_export_special"; + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (name VARCHAR(255) NOT NULL, description VARCHAR(1024), PRIMARY KEY(name))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?)", tableName), + "commas", + "value,with,commas" + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?)", tableName), + "quotes", + "value\"with\"quotes" + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?)", tableName), + "simple", + "plain_value" + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_special_test", ".csv").toFile(); + outputFile.deleteOnExit(); + + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath() + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(3, lines.size()); + Collections.sort(lines); + + // Values with commas should be quoted (sorted order: commas, quotes, simple) + Assert.assertEquals("commas,\"value,with,commas\"", lines.get(0)); + // Values with quotes should be quoted and quotes doubled + Assert.assertEquals("quotes,\"value\"\"with\"\"quotes\"", lines.get(1)); + // Simple values should not be quoted + Assert.assertEquals("simple,plain_value", lines.get(2)); + + dropTable(tableName); + } + + @Test + public void testExportTableWithNullValues() throws IOException + { + final String tableName = "test_export_nulls"; + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (name VARCHAR(255) NOT NULL, payload BLOB, description VARCHAR(255), PRIMARY KEY(name))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?)", tableName), + "with_values", + StringUtils.toUtf8("{\"key\":1}"), + "has_desc" + ); + handle.execute( + StringUtils.format("INSERT INTO %s (name) VALUES (?)", tableName), + "null_cols" + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_nulls_test", ".csv").toFile(); + outputFile.deleteOnExit(); + + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath() + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(2, lines.size()); + Collections.sort(lines); + + // Row with NULL payload and NULL description should have empty fields + Assert.assertEquals("null_cols,,", lines.get(0)); + + // Row with values + final String expectedHex = BaseEncoding.base16().encode(StringUtils.toUtf8("{\"key\":1}")); + Assert.assertEquals("with_values," + expectedHex + ",has_desc", lines.get(1)); + + dropTable(tableName); + } + + @Test + public void testExportTablePreservesAllColumns() throws IOException + { + final String tableName = "test_export_allcols"; + connector.getDBI().withHandle( + handle -> { + // Simulate segments table structure with columns after payload + handle.execute( + StringUtils.format( + "CREATE TABLE %s (" + + "id VARCHAR(255) NOT NULL, " + + "used BOOLEAN NOT NULL, " + + "payload BLOB NOT NULL, " + + "used_status_last_updated VARCHAR(255), " + + "fingerprint VARCHAR(255), " + + "PRIMARY KEY(id))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?, ?, ?)", tableName), + "seg1", + true, + StringUtils.toUtf8("{\"v\":1}"), + "2024-01-01", + "fp_abc" + ); + handle.execute( + StringUtils.format("INSERT INTO %s (id, used, payload) VALUES (?, ?, ?)", tableName), + "seg2", + false, + StringUtils.toUtf8("{\"v\":2}") + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_allcols_test", ".csv").toFile(); + outputFile.deleteOnExit(); + + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath() + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(2, lines.size()); + Collections.sort(lines); + + // All 5 columns should be present, including those after payload + final String hex1 = BaseEncoding.base16().encode(StringUtils.toUtf8("{\"v\":1}")); + Assert.assertEquals("seg1,true," + hex1 + ",2024-01-01,fp_abc", lines.get(0)); + + // NULL trailing columns should produce empty fields + final String hex2 = BaseEncoding.base16().encode(StringUtils.toUtf8("{\"v\":2}")); + Assert.assertEquals("seg2,false," + hex2 + ",,", lines.get(1)); + + dropTable(tableName); + } + static class TestSQLMetadataConnector extends SQLMetadataConnector { public TestSQLMetadataConnector( diff --git a/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java b/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java index 72768e8bce25..e3af4ae9be9a 100644 --- a/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java +++ b/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java @@ -125,6 +125,16 @@ public void tearDown() } } + /** + * Calls the generic JDBC exportTable implementation from {@link SQLMetadataConnector}, + * bypassing Derby's native SYSCS_EXPORT_TABLE override. + * This exercises the same code path used by PostgreSQL and other connectors. + */ + public void exportTableGeneric(final String tableName, final String outputPath) + { + exportTableWithJdbc(tableName, outputPath); + } + public static String dbSafeUUID() { return StringUtils.removeChar(UUID.randomUUID().toString(), '-'); diff --git a/services/src/main/java/org/apache/druid/cli/ExportMetadata.java b/services/src/main/java/org/apache/druid/cli/ExportMetadata.java index f6c6e510f641..cc262cfecb75 100644 --- a/services/src/main/java/org/apache/druid/cli/ExportMetadata.java +++ b/services/src/main/java/org/apache/druid/cli/ExportMetadata.java @@ -61,7 +61,7 @@ @Command( name = "export-metadata", - description = "Exports the contents of a Druid Derby metadata store to CSV files to assist with cluster migration. This tool also provides the ability to rewrite segment locations in the Derby metadata to assist with deep storage migration." + description = "Exports the contents of a Druid metadata store (Derby or PostgreSQL) to CSV files to assist with cluster migration. This tool also provides the ability to rewrite segment locations in the metadata to assist with deep storage migration." ) public class ExportMetadata extends GuiceRunnable { @@ -245,12 +245,18 @@ private void exportTable( } else { pathFormatString = "%s/%s.csv"; } + final String exportTableName = isDerby() ? StringUtils.toUpperCase(tableName) : tableName; dbConnector.exportTable( - StringUtils.toUpperCase(tableName), + exportTableName, StringUtils.format(pathFormatString, outputPath, tableName) ); } + private boolean isDerby() + { + return connectURI != null && connectURI.startsWith("jdbc:derby"); + } + private void rewriteDatasourceExport( String datasourceTableName ) @@ -267,10 +273,10 @@ private void rewriteDatasourceExport( while ((line = reader.readLine()) != null) { String[] parsed = PARSER.parseLine(line); - String newLine = parsed[0] + "," //dataSource - + parsed[1] + "," //created_date + String newLine = csvEscapeField(parsed[0]) + "," //dataSource + + csvEscapeField(parsed[1]) + "," //created_date + rewriteHexPayloadAsEscapedJson(parsed[2]) + "," //commit_metadata_payload - + parsed[3] //commit_metadata_sha1 + + csvEscapeField(parsed[3]) //commit_metadata_sha1 + "\n"; writer.write(newLine); @@ -297,9 +303,9 @@ private void rewriteRulesExport( while ((line = reader.readLine()) != null) { String[] parsed = PARSER.parseLine(line); - String newLine = parsed[0] + "," //id - + parsed[1] + "," //dataSource - + parsed[2] + "," //version + String newLine = csvEscapeField(parsed[0]) + "," //id + + csvEscapeField(parsed[1]) + "," //dataSource + + csvEscapeField(parsed[2]) + "," //version + rewriteHexPayloadAsEscapedJson(parsed[3]) //payload + "\n"; writer.write(newLine); @@ -327,7 +333,7 @@ private void rewriteConfigExport( while ((line = reader.readLine()) != null) { String[] parsed = PARSER.parseLine(line); - String newLine = parsed[0] + "," //name + String newLine = csvEscapeField(parsed[0]) + "," //name + rewriteHexPayloadAsEscapedJson(parsed[1]) //payload + "\n"; writer.write(newLine); @@ -355,9 +361,9 @@ private void rewriteSupervisorExport( while ((line = reader.readLine()) != null) { String[] parsed = PARSER.parseLine(line); - String newLine = parsed[0] + "," //id - + parsed[1] + "," //spec_id - + parsed[2] + "," //created_date + String newLine = csvEscapeField(parsed[0]) + "," //id + + csvEscapeField(parsed[1]) + "," //spec_id + + csvEscapeField(parsed[2]) + "," //created_date + rewriteHexPayloadAsEscapedJson(parsed[3]) //payload + "\n"; writer.write(newLine); @@ -370,7 +376,7 @@ private void rewriteSupervisorExport( } - private void rewriteSegmentsExport( + void rewriteSegmentsExport( String segmentsTableName ) { @@ -386,13 +392,13 @@ private void rewriteSegmentsExport( while ((line = reader.readLine()) != null) { String[] parsed = PARSER.parseLine(line); StringBuilder newLineBuilder = new StringBuilder(); - newLineBuilder.append(parsed[0]).append(","); //id - newLineBuilder.append(parsed[1]).append(","); //dataSource - newLineBuilder.append(parsed[2]).append(","); //created_date - newLineBuilder.append(parsed[3]).append(","); //start - newLineBuilder.append(parsed[4]).append(","); //end + newLineBuilder.append(csvEscapeField(parsed[0])).append(","); //id + newLineBuilder.append(csvEscapeField(parsed[1])).append(","); //dataSource + newLineBuilder.append(csvEscapeField(parsed[2])).append(","); //created_date + newLineBuilder.append(csvEscapeField(parsed[3])).append(","); //start + newLineBuilder.append(csvEscapeField(parsed[4])).append(","); //end newLineBuilder.append(convertBooleanString(parsed[5])).append(","); //partitioned - newLineBuilder.append(parsed[6]).append(","); //version + newLineBuilder.append(csvEscapeField(parsed[6])).append(","); //version newLineBuilder.append(convertBooleanString(parsed[7])).append(","); //used if (s3Bucket != null || hadoopStorageDirectory != null || newLocalPath != null) { @@ -400,6 +406,13 @@ private void rewriteSegmentsExport( } else { newLineBuilder.append(rewriteHexPayloadAsEscapedJson(parsed[8])); //payload } + + // Preserve any additional columns after payload (e.g. used_status_last_updated, + // indexing_state_fingerprint, upgraded_from_segment_id, schema_fingerprint, num_rows) + for (int i = 9; i < parsed.length; i++) { + newLineBuilder.append(","); + newLineBuilder.append(csvEscapeField(parsed[i])); + } newLineBuilder.append("\n"); writer.write(newLineBuilder.toString()); @@ -473,6 +486,22 @@ private String escapeJSONForCSV(String json) return "\"" + StringUtils.replace(json, "\"", "\"\"") + "\""; } + /** + * Escapes a field value for CSV output following RFC 4180. + * If the value contains commas, double quotes, or newlines, it is wrapped + * in double quotes with internal double quotes escaped by doubling. + */ + static String csvEscapeField(String value) + { + if (value == null) { + return ""; + } + if (value.contains(",") || value.contains("\"") || value.contains("\n") || value.contains("\r")) { + return "\"" + StringUtils.replace(value, "\"", "\"\"") + "\""; + } + return value; + } + private Map makeS3LoadSpec( String segmentPath ) diff --git a/services/src/test/java/org/apache/druid/cli/ExportMetadataTest.java b/services/src/test/java/org/apache/druid/cli/ExportMetadataTest.java new file mode 100644 index 000000000000..763f5a3126bb --- /dev/null +++ b/services/src/test/java/org/apache/druid/cli/ExportMetadataTest.java @@ -0,0 +1,261 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.cli; + +import com.google.common.io.BaseEncoding; +import com.opencsv.CSVParser; +import org.apache.druid.java.util.common.StringUtils; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.List; + +public class ExportMetadataTest +{ + @Rule + public final TemporaryFolder tempFolder = new TemporaryFolder(); + + @Test + public void testCsvEscapeField_plainValue() + { + Assert.assertEquals("hello", ExportMetadata.csvEscapeField("hello")); + } + + @Test + public void testCsvEscapeField_withComma() + { + Assert.assertEquals("\"value,with,commas\"", ExportMetadata.csvEscapeField("value,with,commas")); + } + + @Test + public void testCsvEscapeField_withDoubleQuote() + { + Assert.assertEquals("\"value\"\"with\"\"quotes\"", ExportMetadata.csvEscapeField("value\"with\"quotes")); + } + + @Test + public void testCsvEscapeField_withNewline() + { + Assert.assertEquals("\"line1\nline2\"", ExportMetadata.csvEscapeField("line1\nline2")); + } + + @Test + public void testCsvEscapeField_withCarriageReturn() + { + Assert.assertEquals("\"line1\rline2\"", ExportMetadata.csvEscapeField("line1\rline2")); + } + + @Test + public void testCsvEscapeField_null() + { + Assert.assertEquals("", ExportMetadata.csvEscapeField(null)); + } + + @Test + public void testCsvEscapeField_empty() + { + Assert.assertEquals("", ExportMetadata.csvEscapeField("")); + } + + @Test + public void testRewriteSegmentsExport_preservesAllColumns() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_export"); + final String tableName = "druid_segments"; + + // Build a raw CSV with 12 columns matching the current segments table schema: + // id, dataSource, created_date, start, end, partitioned, version, used, payload, + // used_status_last_updated, indexing_state_fingerprint, upgraded_from_segment_id + final String payloadJson = "{\"type\":\"test\"}"; + final String payloadHex = BaseEncoding.base16().encode(StringUtils.toUtf8(payloadJson)); + + final String rawLine = String.join(",", + "seg_id_1", + "my_datasource", + "2024-01-15", + "2024-01-01", + "2024-01-02", + "true", + "v1", + "true", + payloadHex, + "2024-06-01T00:00:00.000Z", + "fp_abc123", + "upgraded_seg_0" + ); + + final File rawFile = new File(outputDir, tableName + "_raw.csv"); + try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(rawFile), StandardCharsets.UTF_8)) { + writer.write(rawLine + "\n"); + } + + final ExportMetadata exporter = new ExportMetadata(); + exporter.outputPath = outputDir.getAbsolutePath(); + exporter.useHexBlobs = false; + exporter.booleansAsStrings = false; + + exporter.rewriteSegmentsExport(tableName); + + final File outFile = new File(outputDir, tableName + ".csv"); + Assert.assertTrue("Output CSV must exist", outFile.exists()); + + final List lines = Files.readAllLines(outFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(1, lines.size()); + + // Parse the output with opencsv to verify field count and values + final CSVParser parser = new CSVParser(); + final String[] fields = parser.parseLine(lines.get(0)); + + // Must have all 12 columns + Assert.assertEquals("All 12 columns must be preserved", 12, fields.length); + + Assert.assertEquals("seg_id_1", fields[0]); + Assert.assertEquals("my_datasource", fields[1]); + Assert.assertEquals("2024-01-15", fields[2]); + Assert.assertEquals("2024-01-01", fields[3]); + Assert.assertEquals("2024-01-02", fields[4]); + Assert.assertEquals("1", fields[5]); // partitioned: true -> 1 + Assert.assertEquals("v1", fields[6]); + Assert.assertEquals("1", fields[7]); // used: true -> 1 + + // payload should be escaped JSON, not hex + Assert.assertEquals(payloadJson, fields[8]); + + // Additional columns preserved + Assert.assertEquals("2024-06-01T00:00:00.000Z", fields[9]); + Assert.assertEquals("fp_abc123", fields[10]); + Assert.assertEquals("upgraded_seg_0", fields[11]); + } + + @Test + public void testRewriteSegmentsExport_withSpecialCharsInFields() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_special"); + final String tableName = "druid_segments"; + + final String payloadJson = "{\"type\":\"test\"}"; + final String payloadHex = BaseEncoding.base16().encode(StringUtils.toUtf8(payloadJson)); + + // datasource with comma, version with quotes — these need proper CSV escaping + final String datasource = "ds,with,commas"; + final String version = "v\"quoted\""; + + // Column order: id, dataSource, created_date, start, end, partitioned, version, used, payload, ... + final String rawLineCorrected = csvEscapeField("seg,id,1") + "," + + csvEscapeField(datasource) + "," + + "2024-01-15," + + "2024-01-01," + + "2024-01-02," + + "true," + + csvEscapeField(version) + "," + + "false," + + payloadHex + "," + + "2024-06-01"; + + final File rawFile = new File(outputDir, tableName + "_raw.csv"); + try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(rawFile), StandardCharsets.UTF_8)) { + writer.write(rawLineCorrected + "\n"); + } + + final ExportMetadata exporter = new ExportMetadata(); + exporter.outputPath = outputDir.getAbsolutePath(); + exporter.useHexBlobs = false; + exporter.booleansAsStrings = false; + + exporter.rewriteSegmentsExport(tableName); + + final File outFile = new File(outputDir, tableName + ".csv"); + final List lines = Files.readAllLines(outFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(1, lines.size()); + + // Parse output and verify special characters survived the round-trip + final CSVParser parser = new CSVParser(); + final String[] fields = parser.parseLine(lines.get(0)); + + Assert.assertEquals(10, fields.length); + Assert.assertEquals("seg,id,1", fields[0]); + Assert.assertEquals(datasource, fields[1]); + Assert.assertEquals(version, fields[6]); + Assert.assertEquals("2024-06-01", fields[9]); + } + + @Test + public void testRewriteSegmentsExport_with9ColumnsOnly() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_9cols"); + final String tableName = "druid_segments"; + + // Simulate an older segments table that only has 9 columns (no used_status_last_updated, etc.) + final String payloadJson = "{\"type\":\"old\"}"; + final String payloadHex = BaseEncoding.base16().encode(StringUtils.toUtf8(payloadJson)); + + final String rawLine = String.join(",", + "old_seg", + "old_ds", + "2020-01-01", + "2020-01-01", + "2020-01-02", + "false", + "v0", + "true", + payloadHex + ); + + final File rawFile = new File(outputDir, tableName + "_raw.csv"); + try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(rawFile), StandardCharsets.UTF_8)) { + writer.write(rawLine + "\n"); + } + + final ExportMetadata exporter = new ExportMetadata(); + exporter.outputPath = outputDir.getAbsolutePath(); + exporter.useHexBlobs = false; + exporter.booleansAsStrings = false; + + exporter.rewriteSegmentsExport(tableName); + + final File outFile = new File(outputDir, tableName + ".csv"); + final List lines = Files.readAllLines(outFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(1, lines.size()); + + final CSVParser parser = new CSVParser(); + final String[] fields = parser.parseLine(lines.get(0)); + + // Should still work with only 9 columns + Assert.assertEquals(9, fields.length); + Assert.assertEquals("old_seg", fields[0]); + Assert.assertEquals(payloadJson, fields[8]); + } + + /** + * Local helper matching ExportMetadata.csvEscapeField for building test input. + */ + private static String csvEscapeField(String value) + { + return ExportMetadata.csvEscapeField(value); + } +}