FrankChen021 commented on code in PR #19698:
URL: https://github.com/apache/druid/pull/19698#discussion_r3880823861
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java:
##########
@@ -1055,6 +1062,194 @@ 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<String> 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 returned by {@link
#getMetadataTableSchema(Connection)}, which is the schema
+ * an unqualified table name resolves to. Rather than passing the table name
as a search pattern, in which '_' is a
+ * wildcard and which is case-sensitive while the database folds unquoted
identifiers, the returned table names are
+ * compared to the given one ignoring case.
+ */
+ public List<String> getTableColumns(final String tableName)
+ {
+ return getDBI().withHandle(handle -> {
+ final List<String> columns = new ArrayList<>();
+ try {
+ if (tableExists(handle, tableName)) {
+ final Connection conn = handle.getConnection();
+ try (ResultSet rs = conn.getMetaData().getColumns(null,
getMetadataTableSchema(conn), null, null)) {
Review Comment:
[P2] Schema lookup still treats the schema as a pattern
getColumns receives getMetadataTableSchema(conn) as its schema-pattern
argument. JDBC metadata patterns treat '_' and '%' as wildcards, so a
configured schema containing either character can return columns from other
schemas; the later TABLE_NAME check does not verify TABLE_SCHEM. Escape the
schema pattern or filter TABLE_SCHEM for an exact match.
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java:
##########
@@ -1055,6 +1062,194 @@ 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<String> 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 returned by {@link
#getMetadataTableSchema(Connection)}, which is the schema
+ * an unqualified table name resolves to. Rather than passing the table name
as a search pattern, in which '_' is a
+ * wildcard and which is case-sensitive while the database folds unquoted
identifiers, the returned table names are
+ * compared to the given one ignoring case.
+ */
+ public List<String> getTableColumns(final String tableName)
+ {
+ return getDBI().withHandle(handle -> {
+ final List<String> columns = new ArrayList<>();
+ try {
+ if (tableExists(handle, tableName)) {
+ final Connection conn = handle.getConnection();
+ try (ResultSet rs = conn.getMetaData().getColumns(null,
getMetadataTableSchema(conn), null, null)) {
+ while (rs.next()) {
+ if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME"))) {
+ columns.add(rs.getString("COLUMN_NAME"));
+ }
+ }
+ }
+ }
+ }
+ catch (SQLException e) {
+ log.warn(e, "Could not read columns of table[%s].", tableName);
+ }
+ return columns;
+ });
+ }
+
+ /**
+ * Returns the schema that the Druid metadata tables live in, i.e. the
schema that an unqualified
+ * table name in a Druid SQL statement resolves to, or null if the schema is
unknown and lookups
+ * should not be scoped to a schema.
+ *
+ * Connectors that scope {@link #tableExists} to a configured schema must
override this so that
+ * both lookups agree.
+ */
+ @Nullable
+ protected String getMetadataTableSchema(final Connection connection) throws
SQLException
+ {
+ return connection.getSchema();
+ }
+
+ /**
+ * 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<String> columns) throws SQLException
+ {
+ final String quote = conn.getMetaData().getIdentifierQuoteString();
+ return columns.stream()
+ .map(column -> quoteIdentifier(quote, column))
+ .collect(Collectors.joining(","));
+ }
+
+ /**
+ * Quotes an identifier with the given identifier quote string of the
database, doubling any occurrence of the quote
+ * string inside the identifier. Returns the identifier unchanged if the
database does not support quoting, which
+ * {@link DatabaseMetaData#getIdentifierQuoteString()} reports as a space.
+ */
+ private static String quoteIdentifier(@Nullable final String quote, final
String identifier)
+ {
+ if (quote == null || " ".equals(quote)) {
+ return identifier;
+ }
+ return quote + StringUtils.replace(identifier, quote, quote + quote) +
quote;
+ }
+
+ /**
+ * 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<String> 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<Void>) (handle, status) -> {
+ final Connection conn = handle.getConnection();
+ final String selectList = columns == null || columns.isEmpty() ? "*"
: makeExportSelectList(conn, columns);
+ // Qualify the table with the schema that Druid's tables live in,
which is not necessarily the schema an
+ // unqualified name resolves to for this connection. The schema is
quoted, since it is the name as stored in
+ // the database, while the table name is left unquoted so that it is
folded by the database in the same way
+ // as in every other Druid statement.
+ final String schema = getMetadataTableSchema(conn);
+ final String qualifiedTableName =
+ schema == null
+ ? tableName
+ : quoteIdentifier(conn.getMetaData().getIdentifierQuoteString(),
schema) + "." + tableName;
+ try (Statement stmt = conn.createStatement()) {
+ // Set the fetch size unconditionally: some drivers use a sentinel
value to request streaming, such as
+ // Integer.MIN_VALUE in MySQL, which would be discarded by a
positive-value check.
+ stmt.setFetchSize(getStreamingFetchSize());
+ try (ResultSet rs = stmt.executeQuery(
+ StringUtils.format("SELECT %s FROM %s", selectList,
qualifiedTableName)
+ );
+ OutputStreamWriter writer =
+ new OutputStreamWriter(new FileOutputStream(outputPath),
StandardCharsets.UTF_8)) {
+ final int columnCount = rs.getMetaData().getColumnCount();
+ final List<String> values = new ArrayList<>(columnCount);
+ while (rs.next()) {
+ values.clear();
+ for (int i = 1; i <= columnCount; i++) {
+ values.add(readCsvValue(rs, i));
+ }
+ writer.write(String.join(",", values));
+ writer.write('\n');
+ }
+ }
+ }
+ return null;
+ },
+ QUIET_RETRIES,
+ DEFAULT_MAX_TRIES
+ );
+ }
+
+ /**
+ * Reads the given column of the current row as a CSV field. Binary values
are hex-encoded, booleans are written
+ * as true/false and NULLs as empty fields.
+ */
+ private static String readCsvValue(final ResultSet rs, final int column)
throws SQLException
+ {
+ final ResultSetMetaData meta = rs.getMetaData();
+ final int type = meta.getColumnType(column);
+ if (type == Types.BINARY || type == Types.VARBINARY || type ==
Types.LONGVARBINARY || type == Types.BLOB
+ || (type == Types.OTHER &&
"bytea".equalsIgnoreCase(meta.getColumnTypeName(column)))) {
+ final byte[] bytes = rs.getBytes(column);
+ return bytes == null ? "" : BaseEncoding.base16().encode(bytes);
+ } else if (type == Types.BOOLEAN || type == Types.BIT) {
+ final boolean value = rs.getBoolean(column);
+ return rs.wasNull() ? "" : String.valueOf(value);
+ } else {
+ return csvEscape(rs.getString(column));
+ }
+ }
+
+ /**
+ * Escapes a value for CSV output as per RFC 4180: values containing a
comma, double quote or line break are
+ * wrapped in double quotes, with inner double quotes doubled. A null value
is written as an empty field.
+ */
+ public static String csvEscape(@Nullable final String value)
+ {
+ if (value == null) {
+ return "";
+ } else if (value.contains(",") || value.contains("\"") ||
value.contains("\n") || value.contains("\r")) {
Review Comment:
[P2] Empty strings are exported as NULL fields
csvEscape(null) and csvEscape("") both produce an unquoted empty field.
PostgreSQL CSV interprets an unquoted empty field as an empty string, while the
import guidance and FORCE_NULL columns rely on empty fields representing NULL;
conversely, a real empty metadata value cannot round-trip distinctly from NULL.
Quote empty strings or configure NULL handling explicitly so null and empty
values remain distinct.
##########
docs/operations/export-metadata.md:
##########
@@ -151,46 +164,67 @@ In the example command above:
After running the tool, the output directory will contain
`<table-name>_raw.csv` and `<table-name>.csv` files.
-The `<table-name>_raw.csv` files are intermediate files used by the tool,
containing the table data as exported by Derby without modification.
+The `<table-name>_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 `<table-name>.csv` files are used for import into another database such as
MySQL and PostgreSQL and have any configured deep storage location rewrites
applied.
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 and to the user variables of the MySQL command.
+
+If the source table does not have `used_status_last_updated` but the target
table does, the import fails, because Druid creates that column as `NOT NULL`
without a default. Make the column nullable before importing, and give the
imported rows a value afterwards:
+
+```sql
+ALTER TABLE druid_segments ALTER COLUMN used_status_last_updated NULL;
+-- run the import command for your database, omitting used_status_last_updated
from the column list
+UPDATE druid_segments SET used_status_last_updated = created_date WHERE
used_status_last_updated IS NULL;
+ALTER TABLE druid_segments ALTER COLUMN used_status_last_updated NOT NULL;
+```
+
+The `ALTER TABLE` syntax above is for Derby. On PostgreSQL, use `ALTER COLUMN
used_status_last_updated DROP NOT NULL` and `SET NOT NULL`; on MySQL, use
`MODIFY used_status_last_updated VARCHAR(255) NULL` and `MODIFY
used_status_last_updated VARCHAR(255) NOT NULL`.
+
+NULL values are written as empty fields, and each database needs to be told
how to import them:
+
+- Derby imports an empty field as NULL, so no extra handling is needed.
+- PostgreSQL `COPY` imports an empty field as an empty string, which fails for
non-string columns such as `num_rows`, so the command below declares the
nullable columns with `FORCE_NULL`.
+- MySQL `LOAD DATA` imports an empty field as an empty string, and coerces it
to `0` for numeric columns such as `num_rows`, so the command below reads the
nullable columns into user variables and converts empty values to NULL with
`NULLIF`.
+
+The exported CSV follows RFC 4180, in which a backslash is an ordinary
character. MySQL `LOAD DATA` treats backslashes as escape characters by
default, which would corrupt payloads and segment ids containing them, so the
commands below disable this with `ESCAPED BY ''`.
+
### Derby
```sql
-CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE
(null,'DRUID_SEGMENTS','/tmp/csv/druid_segments.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_SEGMENTS','id,dataSource,created_date,start,"end",partitioned,version,used,payload,used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id',null,'/tmp/csv/druid_segments.csv',',','"',null,0);
-CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE
(null,'DRUID_RULES','/tmp/csv/druid_rules.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_RULES','id,dataSource,version,payload',null,'/tmp/csv/druid_rules.csv',',','"',null,0);
-CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE
(null,'DRUID_CONFIG','/tmp/csv/druid_config.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_CONFIG','name,payload',null,'/tmp/csv/druid_config.csv',',','"',null,0);
-CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE
(null,'DRUID_DATASOURCE','/tmp/csv/druid_dataSource.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_DATASOURCE','dataSource,created_date,commit_metadata_payload,commit_metadata_sha1',null,'/tmp/csv/druid_dataSource.csv',',','"',null,0);
-CALL SYSCS_UTIL.SYSCS_IMPORT_TABLE
(null,'DRUID_SUPERVISORS','/tmp/csv/druid_supervisors.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_SUPERVISORS','id,spec_id,created_date,payload',null,'/tmp/csv/druid_supervisors.csv',',','"',null,0);
```
### 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 '\"' ESCAPED BY ''
(id,dataSource,created_date,start,end,partitioned,version,used,payload,@used_status_last_updated,@indexing_state_fingerprint,@upgraded_from_segment_id)
SET used_status_last_updated=NULLIF(@used_status_last_updated,''),
indexing_state_fingerprint=NULLIF(@indexing_state_fingerprint,''),
upgraded_from_segment_id=NULLIF(@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;
+LOAD DATA INFILE '/tmp/csv/druid_rules.csv' INTO TABLE druid_rules FIELDS
TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' ESCAPED BY ''
(id,dataSource,version,payload); SHOW WARNINGS;
-LOAD DATA INFILE '/tmp/csv/druid_config.csv' INTO TABLE druid_config FIELDS
TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' (name,payload); SHOW WARNINGS;
+LOAD DATA INFILE '/tmp/csv/druid_config.csv' INTO TABLE druid_config FIELDS
TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' ESCAPED BY '' (name,payload);
SHOW WARNINGS;
-LOAD DATA INFILE '/tmp/csv/druid_dataSource.csv' INTO TABLE druid_dataSource
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'
(dataSource,created_date,commit_metadata_payload,commit_metadata_sha1); SHOW
WARNINGS;
+LOAD DATA INFILE '/tmp/csv/druid_dataSource.csv' INTO TABLE druid_dataSource
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' ESCAPED BY ''
(dataSource,created_date,commit_metadata_payload,commit_metadata_sha1); SHOW
WARNINGS;
-LOAD DATA INFILE '/tmp/csv/druid_supervisors.csv' INTO TABLE druid_supervisors
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'
(id,spec_id,created_date,payload); SHOW WARNINGS;
+LOAD DATA INFILE '/tmp/csv/druid_supervisors.csv' INTO TABLE druid_supervisors
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' ESCAPED BY ''
(id,spec_id,created_date,payload); SHOW WARNINGS;
```
### 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));
Review Comment:
[P1] PostgreSQL COPY can corrupt rewritten BYTEA payloads
The PostgreSQL import example uses COPY CSV with backslash escapes enabled,
while the exporter writes BYTEA values as JSON text containing literal
backslashes. PostgreSQL can interpret those backslashes as escape syntax,
causing escaped payloads to fail or change during import. Emit BYTEA as
\x-prefixed hex or use an explicit decode expression so the documented import
round-trips the exported data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
