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..ad145077a181 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. @@ -159,6 +172,10 @@ Example import commands for Derby, MySQL, and PostgreSQL are shown below. These example import commands expect `/tmp/csv` and its contents to be accessible from the server. For other options, such as importing from the client filesystem, please refer to the database's documentation. +The segments table is exported in a fixed column order, independent of the physical column order of the source table: `id`, `dataSource`, `created_date`, `start`, `end`, `partitioned`, `version`, `used`, `payload`, followed by whichever of the optional columns `used_status_last_updated`, `indexing_state_fingerprint`, `upgraded_from_segment_id`, `schema_fingerprint`, and `num_rows` exist in the source table, in that order. Adjust the segments column list in the import commands below to contain exactly the columns of the source table: omit any optional column the source table does not have (segments tables from older Druid versions may have only the first nine columns), and add `schema_fingerprint,num_rows` at the end if the source table has them. Apply the same adjustment to the columns declared with `FORCE_NULL` in the PostgreSQL command. + +NULL values are written as empty fields. An empty field is imported as an empty string rather than NULL, which fails for non-string columns such as `num_rows`, so the PostgreSQL command below declares the nullable columns with `FORCE_NULL`. + ### Derby ```sql @@ -176,7 +193,7 @@ CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE (null,'DRUID_SUPERVISORS','/tmp/csv/druid_sup ### MySQL ```sql -LOAD DATA INFILE '/tmp/csv/druid_segments.csv' INTO TABLE druid_segments FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' (id,dataSource,created_date,start,end,partitioned,version,used,payload); SHOW WARNINGS; +LOAD DATA INFILE '/tmp/csv/druid_segments.csv' INTO TABLE druid_segments FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' (id,dataSource,created_date,start,end,partitioned,version,used,payload,used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id); SHOW WARNINGS; LOAD DATA INFILE '/tmp/csv/druid_rules.csv' INTO TABLE druid_rules FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' (id,dataSource,version,payload); SHOW WARNINGS; @@ -190,7 +207,7 @@ LOAD DATA INFILE '/tmp/csv/druid_supervisors.csv' INTO TABLE druid_supervisors F ### PostgreSQL ```sql -COPY druid_segments(id,dataSource,created_date,start,"end",partitioned,version,used,payload) FROM '/tmp/csv/druid_segments.csv' DELIMITER ',' CSV; +COPY druid_segments(id,dataSource,created_date,start,"end",partitioned,version,used,payload,used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id) FROM '/tmp/csv/druid_segments.csv' WITH (FORMAT csv, FORCE_NULL (used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id)); COPY druid_rules(id,dataSource,version,payload) FROM '/tmp/csv/druid_rules.csv' DELIMITER ',' CSV; 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..aedd5eb916dc 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,206 @@ public void createAuditTable() } } + @Override + public void exportTable( + final String tableName, + final String outputPath + ) + { + exportTable(tableName, outputPath, null); + } + + /** + * Exports a table to a CSV file, emitting the given columns in the given order. + * + * @param columns columns to export in the desired order, or null to export all columns in the + * order reported by the database + */ + public void exportTable( + final String tableName, + final String outputPath, + @Nullable final List columns + ) + { + exportTableWithJdbc(tableName, outputPath, columns); + } + + /** + * Returns the columns of the given table, in the order reported by the database. + * Returns an empty list if the table does not exist or the metadata cannot be read. + * + * The lookup is scoped to the schema of the current connection, which is the schema an unqualified + * table name resolves to. The table name is folded to the case in which the database stores + * unquoted identifiers, and escaped so that it is matched literally rather than as a + * {@link DatabaseMetaData#getColumns} search pattern. + */ + public List getTableColumns(final String tableName) + { + return getDBI().withHandle(handle -> { + final List columns = new ArrayList<>(); + try { + if (tableExists(handle, tableName)) { + final Connection conn = handle.getConnection(); + final DatabaseMetaData dbMetaData = conn.getMetaData(); + try (ResultSet rs = dbMetaData.getColumns( + null, + escapeMetaDataSearchString(dbMetaData, conn.getSchema()), + escapeMetaDataSearchString(dbMetaData, foldIdentifierCase(dbMetaData, tableName)), + null + )) { + while (rs.next()) { + columns.add(rs.getString("COLUMN_NAME")); + } + } + } + } + catch (SQLException e) { + log.warn(e, "Could not read columns of table[%s].", tableName); + } + return columns; + }); + } + + /** + * Folds the given identifier to the case in which the database stores unquoted identifiers. + * {@link DatabaseMetaData} lookup patterns are case-sensitive, while an unquoted identifier + * in a SQL statement is folded by the database (to lowercase in PostgreSQL, to uppercase in + * Derby), so the folded form must be used to match the table a SQL reference resolves to. + */ + private static String foldIdentifierCase( + final DatabaseMetaData dbMetaData, + final String identifier + ) throws SQLException + { + if (dbMetaData.storesLowerCaseIdentifiers()) { + return StringUtils.toLowerCase(identifier); + } + if (dbMetaData.storesUpperCaseIdentifiers()) { + return StringUtils.toUpperCase(identifier); + } + return identifier; + } + + /** + * Escapes the wildcard characters of a {@link DatabaseMetaData} search pattern, so that the given + * value is matched literally. Returns null if the given value is null, which matches any value. + */ + @Nullable + private static String escapeMetaDataSearchString( + final DatabaseMetaData dbMetaData, + @Nullable final String value + ) throws SQLException + { + final String escape = dbMetaData.getSearchStringEscape(); + if (value == null || escape == null || escape.isEmpty()) { + return value; + } + final StringBuilder escaped = new StringBuilder(value.length()); + for (int i = 0; i < value.length(); i++) { + final char c = value.charAt(i); + if (c == '_' || c == '%' || c == escape.charAt(0)) { + escaped.append(escape); + } + escaped.append(c); + } + return escaped.toString(); + } + + /** + * Builds the select list for an export query, quoting each column with the database's + * identifier quote string so that reserved words such as "end" are handled correctly. + */ + protected String makeExportSelectList(final Connection conn, final List columns) throws SQLException + { + String quote = conn.getMetaData().getIdentifierQuoteString(); + if (quote == null || " ".equals(quote)) { + quote = ""; + } + final String quoteString = quote; + return columns.stream() + .map(column -> quoteString.isEmpty() + ? column + : quoteString + StringUtils.replace(column, quoteString, quoteString + quoteString) + + quoteString) + .collect(Collectors.joining(",")); + } + + /** + * 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. + * + * @param columns columns to export in the desired order, or null to export all columns + */ + protected void exportTableWithJdbc( + final String tableName, + final String outputPath, + @Nullable final List columns + ) + { + // 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); + } + final String selectList = + columns == null || columns.isEmpty() ? "*" : makeExportSelectList(conn, columns); + try (ResultSet rs = stmt.executeQuery( + StringUtils.format("SELECT %s FROM %s", selectList, 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/main/java/org/apache/druid/metadata/storage/derby/DerbyConnector.java b/server/src/main/java/org/apache/druid/metadata/storage/derby/DerbyConnector.java index cb4815d2313e..cb86195e8f68 100644 --- a/server/src/main/java/org/apache/druid/metadata/storage/derby/DerbyConnector.java +++ b/server/src/main/java/org/apache/druid/metadata/storage/derby/DerbyConnector.java @@ -36,9 +36,11 @@ import org.skife.jdbi.v2.Handle; import org.skife.jdbi.v2.tweak.HandleCallback; +import javax.annotation.Nullable; import java.sql.DatabaseMetaData; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.List; import java.util.Locale; @ManageLifecycle @@ -131,25 +133,30 @@ public String limitClause(int limit) @Override public void exportTable( String tableName, - String outputPath + String outputPath, + @Nullable List columns ) { retryWithHandle( - new HandleCallback() - { - @Override - public Void withHandle(Handle handle) - { - handle.createStatement( - StringUtils.format( + (HandleCallback) handle -> { + final String statement; + if (columns == null || columns.isEmpty()) { + statement = StringUtils.format( "CALL SYSCS_UTIL.SYSCS_EXPORT_TABLE (null, '%s', '%s', null, null, null)", tableName, outputPath - ) - ).execute(); - return null; - } - } + ); + } else { + statement = StringUtils.format( + "CALL SYSCS_UTIL.SYSCS_EXPORT_QUERY ('SELECT %s FROM %s', '%s', null, null, null)", + makeExportSelectList(handle.getConnection(), columns), + tableName, + outputPath + ); + } + handle.createStatement(statement).execute(); + return null; + } ); } 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..50ae5a271dea 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java @@ -21,9 +21,11 @@ import com.google.common.base.Supplier; import com.google.common.base.Suppliers; +import com.google.common.collect.ImmutableList; 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 +39,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 +469,347 @@ 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); + } + + @Test + public void testExportTableWithExplicitColumnOrder() throws IOException + { + final String tableName = "test_export_colorder"; + connector.getDBI().withHandle( + handle -> { + // "end" is a reserved word, so it must be quoted in the export query + handle.execute( + StringUtils.format( + "CREATE TABLE %s (" + + "id VARCHAR(255) NOT NULL, " + + "used_status_last_updated VARCHAR(255), " + + "\"END\" VARCHAR(255), " + + "used BOOLEAN NOT NULL, " + + "PRIMARY KEY(id))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?, ?)", tableName), + "seg1", + "2024-01-01", + "2024-01-02", + true + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_colorder_test", ".csv").toFile(); + outputFile.deleteOnExit(); + + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath(), + ImmutableList.of("ID", "END", "USED", "USED_STATUS_LAST_UPDATED") + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(ImmutableList.of("seg1,2024-01-02,true,2024-01-01"), lines); + + dropTable(tableName); + } + + @Test + public void testExportTableWithDerbyNativeExport() throws IOException + { + // Exercises DerbyConnector's native export, which is the path used by the export-metadata tool + // when the source is Derby + final String tableName = "test_export_native"; + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (" + + "id VARCHAR(255) NOT NULL, " + + "\"END\" VARCHAR(255), " + + "used BOOLEAN NOT NULL, " + + "payload BLOB NOT NULL, " + + "PRIMARY KEY(id))", + tableName + ) + ); + handle.execute( + StringUtils.format("INSERT INTO %s VALUES (?, ?, ?, ?)", tableName), + "seg1", + "2024-01-02", + true, + StringUtils.toUtf8("{\"v\":1}") + ); + return null; + } + ); + + final File outputFile = Files.createTempFile("export_native_test", ".csv").toFile(); + Assert.assertTrue(outputFile.delete()); + outputFile.deleteOnExit(); + + connector.exportTable( + StringUtils.toUpperCase(tableName), + outputFile.getAbsolutePath(), + ImmutableList.of("ID", "PAYLOAD", "END", "USED") + ); + + final List lines = Files.readAllLines(outputFile.toPath(), StandardCharsets.UTF_8); + Assert.assertEquals(1, lines.size()); + + // Derby quotes every field and writes BLOBs as lowercase hex, as expected by the rewrite stage + final String hex = BaseEncoding.base16().lowerCase().encode(StringUtils.toUtf8("{\"v\":1}")); + Assert.assertEquals( + StringUtils.format("\"seg1\",\"%s\",\"2024-01-02\",\"true\"", hex), + lines.get(0) + ); + + dropTable(tableName); + } + + @Test + public void testGetTableColumns() + { + final String tableName = "test_get_columns"; + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (id VARCHAR(255) NOT NULL, used BOOLEAN NOT NULL, PRIMARY KEY(id))", + tableName + ) + ); + return null; + } + ); + + Assert.assertEquals( + ImmutableList.of("ID", "USED"), + connector.getTableColumns(StringUtils.toUpperCase(tableName)) + ); + // A table name in the wrong case must still resolve: the database folds unquoted identifiers + // (Derby to uppercase, PostgreSQL to lowercase), while the metadata lookup is case-sensitive + Assert.assertEquals( + ImmutableList.of("ID", "USED"), + connector.getTableColumns(StringUtils.toLowerCase(tableName)) + ); + Assert.assertEquals(ImmutableList.of(), connector.getTableColumns("NON_EXISTENT_TABLE")); + + 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..ce9d43d11c34 100644 --- a/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java +++ b/server/src/test/java/org/apache/druid/metadata/TestDerbyConnector.java @@ -38,6 +38,7 @@ import org.skife.jdbi.v2.PreparedBatch; import org.skife.jdbi.v2.exceptions.UnableToObtainConnectionException; +import javax.annotation.Nullable; import java.sql.SQLException; import java.util.Arrays; import java.util.HashSet; @@ -125,6 +126,25 @@ 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) + { + exportTableGeneric(tableName, outputPath, null); + } + + public void exportTableGeneric( + final String tableName, + final String outputPath, + @Nullable final List columns + ) + { + exportTableWithJdbc(tableName, outputPath, columns); + } + 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..1c8b2048304f 100644 --- a/services/src/main/java/org/apache/druid/cli/ExportMetadata.java +++ b/services/src/main/java/org/apache/druid/cli/ExportMetadata.java @@ -29,13 +29,16 @@ import com.google.inject.Injector; import com.google.inject.Key; import com.google.inject.Module; -import com.opencsv.CSVParser; +import com.opencsv.CSVReader; +import com.opencsv.CSVReaderBuilder; +import com.opencsv.RFC4180ParserBuilder; import org.apache.druid.guice.DruidProcessingModule; import org.apache.druid.guice.JsonConfigProvider; import org.apache.druid.guice.QueryRunnerFactoryModule; import org.apache.druid.guice.QueryableModule; import org.apache.druid.guice.annotations.Self; import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.ISE; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.logger.Logger; import org.apache.druid.metadata.MetadataStorageConnectorConfig; @@ -55,13 +58,15 @@ import java.io.InputStreamReader; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; @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 { @@ -120,9 +125,30 @@ public class ExportMetadata extends GuiceRunnable description = "Write boolean values as true/false strings instead of 1/0") public boolean booleansAsStrings = false; - private static final Logger log = new Logger(ExportMetadata.class); + /** + * Canonical order in which the columns of the segments table are exported, matching the import + * commands documented in {@code docs/operations/export-metadata.md}. Columns which do not exist in + * the source table are skipped, and any column not listed here is appended after these, in the + * order reported by the source database. + */ + private static final List SEGMENTS_COLUMN_ORDER = ImmutableList.of( + "id", + "dataSource", + "created_date", + "start", + "end", + "partitioned", + "version", + "used", + "payload", + "used_status_last_updated", + "indexing_state_fingerprint", + "upgraded_from_segment_id", + "schema_fingerprint", + "num_rows" + ); - private static final CSVParser PARSER = new CSVParser(); + private static final Logger log = new Logger(ExportMetadata.class); private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper(); @@ -217,7 +243,7 @@ public void run() rewriteDatasourceExport(metadataStorageTablesConfig.getDataSourceTable()); log.info("Exporting segments table: " + metadataStorageTablesConfig.getSegmentsTable()); - exportTable(dbConnector, metadataStorageTablesConfig.getSegmentsTable(), true); + exportSegmentsTable(dbConnector, metadataStorageTablesConfig.getSegmentsTable()); rewriteSegmentsExport(metadataStorageTablesConfig.getSegmentsTable()); log.info("Exporting rules table: " + metadataStorageTablesConfig.getRulesTable()); @@ -245,12 +271,65 @@ 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) ); } + /** + * Exports the segments table with the columns emitted in {@link #SEGMENTS_COLUMN_ORDER}, so that the + * output does not depend on the physical column order of the source table. + */ + private void exportSegmentsTable( + SQLMetadataConnector dbConnector, + String tableName + ) + { + final String exportTableName = isDerby() ? StringUtils.toUpperCase(tableName) : tableName; + final List columns = orderSegmentsColumns(dbConnector.getTableColumns(exportTableName)); + if (columns.isEmpty()) { + throw new ISE( + "Could not read the columns of table[%s]. Cannot export the segments table in a stable column order.", + exportTableName + ); + } + + dbConnector.exportTable( + exportTableName, + StringUtils.format("%s/%s_raw.csv", outputPath, tableName), + columns + ); + } + + /** + * Orders the given actual column names of the segments table by {@link #SEGMENTS_COLUMN_ORDER}, + * appending any unknown columns at the end in their original order. + */ + static List orderSegmentsColumns(List actualColumns) + { + final Map remaining = new LinkedHashMap<>(); + for (String column : actualColumns) { + remaining.put(StringUtils.toLowerCase(column), column); + } + + final List ordered = new ArrayList<>(actualColumns.size()); + for (String column : SEGMENTS_COLUMN_ORDER) { + final String actualColumn = remaining.remove(StringUtils.toLowerCase(column)); + if (actualColumn != null) { + ordered.add(actualColumn); + } + } + ordered.addAll(remaining.values()); + return ordered; + } + + private boolean isDerby() + { + return connectURI != null && connectURI.startsWith("jdbc:derby"); + } + private void rewriteDatasourceExport( String datasourceTableName ) @@ -258,19 +337,15 @@ private void rewriteDatasourceExport( String inFile = StringUtils.format(("%s/%s_raw.csv"), outputPath, datasourceTableName); String outFile = StringUtils.format("%s/%s.csv", outputPath, datasourceTableName); try ( - BufferedReader reader = new BufferedReader( - new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8) - ); + CSVReader reader = openCsvReader(inFile); OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8) ) { - String line; - while ((line = reader.readLine()) != null) { - String[] parsed = PARSER.parseLine(line); - - String newLine = parsed[0] + "," //dataSource - + parsed[1] + "," //created_date + String[] parsed; + while ((parsed = readRecord(reader, inFile, 4)) != null) { + 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); @@ -288,18 +363,14 @@ private void rewriteRulesExport( String inFile = StringUtils.format(("%s/%s_raw.csv"), outputPath, rulesTableName); String outFile = StringUtils.format("%s/%s.csv", outputPath, rulesTableName); try ( - BufferedReader reader = new BufferedReader( - new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8) - ); + CSVReader reader = openCsvReader(inFile); OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8) ) { - String line; - while ((line = reader.readLine()) != null) { - String[] parsed = PARSER.parseLine(line); - - String newLine = parsed[0] + "," //id - + parsed[1] + "," //dataSource - + parsed[2] + "," //version + String[] parsed; + while ((parsed = readRecord(reader, inFile, 4)) != null) { + String newLine = csvEscapeField(parsed[0]) + "," //id + + csvEscapeField(parsed[1]) + "," //dataSource + + csvEscapeField(parsed[2]) + "," //version + rewriteHexPayloadAsEscapedJson(parsed[3]) //payload + "\n"; writer.write(newLine); @@ -318,16 +389,12 @@ private void rewriteConfigExport( String inFile = StringUtils.format(("%s/%s_raw.csv"), outputPath, configTableName); String outFile = StringUtils.format("%s/%s.csv", outputPath, configTableName); try ( - BufferedReader reader = new BufferedReader( - new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8) - ); + CSVReader reader = openCsvReader(inFile); OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8) ) { - String line; - while ((line = reader.readLine()) != null) { - String[] parsed = PARSER.parseLine(line); - - String newLine = parsed[0] + "," //name + String[] parsed; + while ((parsed = readRecord(reader, inFile, 2)) != null) { + String newLine = csvEscapeField(parsed[0]) + "," //name + rewriteHexPayloadAsEscapedJson(parsed[1]) //payload + "\n"; writer.write(newLine); @@ -346,18 +413,14 @@ private void rewriteSupervisorExport( String inFile = StringUtils.format(("%s/%s_raw.csv"), outputPath, supervisorTableName); String outFile = StringUtils.format("%s/%s.csv", outputPath, supervisorTableName); try ( - BufferedReader reader = new BufferedReader( - new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8) - ); + CSVReader reader = openCsvReader(inFile); OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8) ) { - String line; - while ((line = reader.readLine()) != null) { - String[] parsed = PARSER.parseLine(line); - - String newLine = parsed[0] + "," //id - + parsed[1] + "," //spec_id - + parsed[2] + "," //created_date + String[] parsed; + while ((parsed = readRecord(reader, inFile, 4)) != null) { + 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,29 +433,26 @@ private void rewriteSupervisorExport( } - private void rewriteSegmentsExport( + void rewriteSegmentsExport( String segmentsTableName ) { String inFile = StringUtils.format(("%s/%s_raw.csv"), outputPath, segmentsTableName); String outFile = StringUtils.format("%s/%s.csv", outputPath, segmentsTableName); try ( - BufferedReader reader = new BufferedReader( - new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8) - ); + CSVReader reader = openCsvReader(inFile); OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(outFile), StandardCharsets.UTF_8) ) { - String line; - while ((line = reader.readLine()) != null) { - String[] parsed = PARSER.parseLine(line); + String[] parsed; + while ((parsed = readRecord(reader, inFile, 9)) != null) { 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 +460,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()); @@ -410,6 +477,50 @@ private void rewriteSegmentsExport( } } + /** + * Opens a record-aware CSV reader on the given file. A single CSV record may span multiple physical + * lines if a field value contains a newline, so records must be read with + * {@link CSVReader#readNext()} rather than by reading one line at a time. + * + * The parser follows RFC 4180, matching the output written by the export stage: quotes are doubled and + * backslashes are not escape characters. Carriage returns are kept, so that a carriage return which is + * part of a value is preserved rather than silently dropped. This relies on the raw CSV using LF line + * endings, which is the case for the files written by the export stage. + */ + private static CSVReader openCsvReader(String inFile) throws IOException + { + return new CSVReaderBuilder( + new BufferedReader(new InputStreamReader(new FileInputStream(inFile), StandardCharsets.UTF_8)) + ) + .withCSVParser(new RFC4180ParserBuilder().build()) + .withKeepCarriageReturn(true) + .build(); + } + + /** + * Reads the next record of the given reader, or returns null at the end of the file. + * + * @param minFields minimum number of fields the record must have, to fail with a useful message + * instead of an {@link ArrayIndexOutOfBoundsException} on a malformed file + */ + private static String[] readRecord(CSVReader reader, String inFile, int minFields) throws IOException + { + final String[] record = reader.readNext(); + if (record == null) { + return null; + } + if (record.length < minFields) { + throw new ISE( + "Row[%d] of file[%s] has [%d] fields, expected at least [%d].", + reader.getLinesRead(), + inFile, + record.length, + minFields + ); + } + return record; + } + /** * Returns a new load spec in escaped JSON form, with the new deep storage location if configured. */ @@ -473,6 +584,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..fe2cd6cf9704 --- /dev/null +++ b/services/src/test/java/org/apache/druid/cli/ExportMetadataTest.java @@ -0,0 +1,572 @@ +/* + * 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.collect.ImmutableList; +import com.google.common.io.BaseEncoding; +import com.opencsv.CSVReader; +import com.opencsv.CSVReaderBuilder; +import com.opencsv.ICSVParser; +import com.opencsv.RFC4180ParserBuilder; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.metadata.TestDerbyConnector; +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.ArrayList; +import java.util.Comparator; +import java.util.List; + +public class ExportMetadataTest +{ + @Rule + public final TemporaryFolder tempFolder = new TemporaryFolder(); + + @Rule + public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule = + new TestDerbyConnector.DerbyConnectorRule(); + + @Test + public void testOrderSegmentsColumns_reordersToCanonicalOrder() + { + // Columns as reported by a table where the newer columns were added by ALTER TABLE in arbitrary order + final List actual = ImmutableList.of( + "id", + "dataSource", + "created_date", + "start", + "end", + "partitioned", + "version", + "used", + "payload", + "upgraded_from_segment_id", + "num_rows", + "used_status_last_updated", + "schema_fingerprint", + "indexing_state_fingerprint" + ); + + Assert.assertEquals( + ImmutableList.of( + "id", + "dataSource", + "created_date", + "start", + "end", + "partitioned", + "version", + "used", + "payload", + "used_status_last_updated", + "indexing_state_fingerprint", + "upgraded_from_segment_id", + "schema_fingerprint", + "num_rows" + ), + ExportMetadata.orderSegmentsColumns(actual) + ); + } + + @Test + public void testOrderSegmentsColumns_ignoresCaseAndSkipsMissingColumns() + { + final List actual = ImmutableList.of( + "PAYLOAD", + "USED", + "ID", + "DATASOURCE", + "CREATED_DATE", + "START", + "END", + "PARTITIONED", + "VERSION" + ); + + Assert.assertEquals( + ImmutableList.of( + "ID", + "DATASOURCE", + "CREATED_DATE", + "START", + "END", + "PARTITIONED", + "VERSION", + "USED", + "PAYLOAD" + ), + ExportMetadata.orderSegmentsColumns(actual) + ); + } + + @Test + public void testOrderSegmentsColumns_appendsUnknownColumnsAtEnd() + { + final List actual = ImmutableList.of("custom_col", "id", "payload", "another_col"); + + Assert.assertEquals( + ImmutableList.of("id", "payload", "custom_col", "another_col"), + ExportMetadata.orderSegmentsColumns(actual) + ); + } + + @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 ICSVParser parser = new RFC4180ParserBuilder().build(); + 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 ICSVParser parser = new RFC4180ParserBuilder().build(); + 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 ICSVParser parser = new RFC4180ParserBuilder().build(); + 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]); + } + + @Test + public void testRewriteSegmentsExport_preservesBackslashes() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_backslash"); + final String tableName = "druid_segments"; + + // Backslashes are valid in segment ids and datasource names, and must not be treated as CSV escapes + final String id = "foo\\bar_2024-01-01T00:00:00.000Z_2024-01-02T00:00:00.000Z_v1"; + final String datasource = "foo\\bar"; + final String version = "v\\1"; + final String upgradedFrom = "back\\slash,and\"quote"; + + final String payloadJson = "{\"type\":\"test\",\"path\":\"C:\\\\druid\\\\segments\"}"; + final String payloadHex = BaseEncoding.base16().encode(StringUtils.toUtf8(payloadJson)); + + final String rawLine = csvEscapeField(id) + "," + + csvEscapeField(datasource) + "," + + "2024-01-15," + + "2024-01-01," + + "2024-01-02," + + "true," + + csvEscapeField(version) + "," + + "true," + + payloadHex + "," + + "2024-06-01," + + "fp\\123," + + csvEscapeField(upgradedFrom); + + 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 ICSVParser parser = new RFC4180ParserBuilder().build(); + final String[] fields = parser.parseLine(lines.get(0)); + + Assert.assertEquals(12, fields.length); + Assert.assertEquals(id, fields[0]); + Assert.assertEquals(datasource, fields[1]); + Assert.assertEquals(version, fields[6]); + Assert.assertEquals(payloadJson, fields[8]); + Assert.assertEquals("fp\\123", fields[10]); + Assert.assertEquals(upgradedFrom, fields[11]); + } + + /** + * End-to-end test: export a segments table containing values with embedded newlines and carriage + * returns via the generic JDBC export, then rewrite the raw CSV and verify that every record and + * field survived. A record with an embedded newline spans multiple physical lines, so the rewrite + * must read records rather than lines. + */ + @Test + public void testExportAndRewriteSegments_withMultilineFields() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_multiline_e2e"); + final String tableName = "druid_segments"; + + final String multilineDatasource = "ds\nwith\nnewlines"; + final String crDatasource = "ds\rwith\rcarriage"; + final String multilineFingerprint = "line1\nline2,line3"; + final String payloadJson = "{\"type\":\"test\",\"desc\":\"has\\na newline\"}"; + + final TestDerbyConnector connector = derbyConnectorRule.getConnector(); + connector.getDBI().withHandle( + handle -> { + handle.execute( + StringUtils.format( + "CREATE TABLE %s (" + + "id VARCHAR(255) NOT NULL, " + + "dataSource VARCHAR(255) NOT NULL, " + + "created_date VARCHAR(255) NOT NULL, " + + "start VARCHAR(255) NOT NULL, " + + "\"END\" VARCHAR(255) NOT NULL, " + + "partitioned BOOLEAN NOT NULL, " + + "version VARCHAR(255) NOT NULL, " + + "used BOOLEAN NOT NULL, " + + "payload BLOB NOT NULL, " + + "used_status_last_updated VARCHAR(255), " + + "indexing_state_fingerprint VARCHAR(255), " + + "PRIMARY KEY(id))", + tableName + ) + ); + final String insert = StringUtils.format( + "INSERT INTO %s VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + tableName + ); + handle.execute( + insert, + "seg1", + multilineDatasource, + "2024-01-15", + "2024-01-01", + "2024-01-02", + true, + "v1", + true, + StringUtils.toUtf8(payloadJson), + "2024-06-01", + multilineFingerprint + ); + // indexing_state_fingerprint is left NULL for this row + handle.execute( + StringUtils.format( + "INSERT INTO %s (id, dataSource, created_date, start, \"END\", partitioned, version, used, payload, " + + "used_status_last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + tableName + ), + "seg2", + crDatasource, + "2024-01-16", + "2024-01-03", + "2024-01-04", + false, + "v2", + false, + StringUtils.toUtf8(payloadJson), + "2024-06-02" + ); + return null; + } + ); + + connector.exportTableGeneric( + StringUtils.toUpperCase(tableName), + new File(outputDir, tableName + "_raw.csv").getAbsolutePath(), + ImmutableList.of( + "ID", + "DATASOURCE", + "CREATED_DATE", + "START", + "END", + "PARTITIONED", + "VERSION", + "USED", + "PAYLOAD", + "USED_STATUS_LAST_UPDATED", + "INDEXING_STATE_FINGERPRINT" + ) + ); + + 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 records = new ArrayList<>(); + try (CSVReader reader = new CSVReaderBuilder( + Files.newBufferedReader(outFile.toPath(), StandardCharsets.UTF_8)) + .withCSVParser(new RFC4180ParserBuilder().build()) + .withKeepCarriageReturn(true) + .build()) { + String[] record; + while ((record = reader.readNext()) != null) { + records.add(record); + } + } + + records.sort(Comparator.comparing(record -> record[0])); + Assert.assertEquals(2, records.size()); + + final String[] seg1 = records.get(0); + Assert.assertEquals(11, seg1.length); + Assert.assertEquals("seg1", seg1[0]); + Assert.assertEquals(multilineDatasource, seg1[1]); + Assert.assertEquals("1", seg1[5]); + Assert.assertEquals("1", seg1[7]); + Assert.assertEquals(payloadJson, seg1[8]); + Assert.assertEquals("2024-06-01", seg1[9]); + Assert.assertEquals(multilineFingerprint, seg1[10]); + + final String[] seg2 = records.get(1); + Assert.assertEquals(11, seg2.length); + Assert.assertEquals("seg2", seg2[0]); + Assert.assertEquals(crDatasource, seg2[1]); + Assert.assertEquals("0", seg2[5]); + Assert.assertEquals("0", seg2[7]); + Assert.assertEquals(payloadJson, seg2[8]); + Assert.assertEquals("", seg2[10]); + } + + @Test + public void testRewriteSegmentsExport_failsOnTruncatedRow() throws IOException + { + final File outputDir = tempFolder.newFolder("segments_truncated"); + final String tableName = "druid_segments"; + + final File rawFile = new File(outputDir, tableName + "_raw.csv"); + try (OutputStreamWriter writer = new OutputStreamWriter(new FileOutputStream(rawFile), StandardCharsets.UTF_8)) { + writer.write("only_id,only_datasource\n"); + } + + final ExportMetadata exporter = new ExportMetadata(); + exporter.outputPath = outputDir.getAbsolutePath(); + exporter.useHexBlobs = false; + exporter.booleansAsStrings = false; + + final ISE e = Assert.assertThrows(ISE.class, () -> exporter.rewriteSegmentsExport(tableName)); + Assert.assertTrue(e.getMessage(), e.getMessage().contains("has [2] fields, expected at least [9]")); + } + + /** + * Local helper matching ExportMetadata.csvEscapeField for building test input. + */ + private static String csvEscapeField(String value) + { + return ExportMetadata.csvEscapeField(value); + } +}