FrankChen021 commented on code in PR #19698:
URL: https://github.com/apache/druid/pull/19698#discussion_r3749656028
##########
docs/operations/export-metadata.md:
##########
@@ -151,46 +164,56 @@ 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.
+
+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);
Review Comment:
[P1] Legacy 9-column imports fail against current targets
Omitting used_status_last_updated from a legacy 9-column import leaves the
current target's NOT NULL column unset, causing Derby/PostgreSQL imports to
fail. Document a target-compatible default/value or make the migration create
that column nullable before import.
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java:
##########
@@ -1054,6 +1061,181 @@ 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();
+ if (quote == null || " ".equals(quote)) {
+ return String.join(",", columns);
+ }
+ return columns.stream()
+ .map(column -> quote + StringUtils.replace(column, quote,
quote + quote) + quote)
+ .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<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 identifiers
are left unquoted so that they are
+ // 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 :
schema + "." + tableName;
+ try (Statement stmt = conn.createStatement()) {
+ final int fetchSize = getStreamingFetchSize();
+ if (fetchSize > 0) {
Review Comment:
[P2] MySQL streaming sentinel is discarded
MySQLConnector returns Integer.MIN_VALUE as its streaming sentinel, but the
fetchSize > 0 guard discards it. Large exports then buffer results instead of
streaming and can exhaust memory. Preserve connector-specific sentinel values
or add a dedicated streaming hook.
##########
server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java:
##########
@@ -1054,6 +1061,181 @@ 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();
+ if (quote == null || " ".equals(quote)) {
+ return String.join(",", columns);
+ }
+ return columns.stream()
+ .map(column -> quote + StringUtils.replace(column, quote,
quote + quote) + quote)
+ .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<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 identifiers
are left unquoted so that they are
+ // 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 :
schema + "." + tableName;
Review Comment:
[P2] Configured PostgreSQL schema names are interpolated unquoted
A valid quoted schema such as MetaData is discovered correctly but is then
queried as lowercase metadata, so export fails on case-sensitive PostgreSQL
schemas. Quote the schema identifier using the JDBC driver's identifier-quote
mechanism.
--
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]
