feat: Support PostgreSQL export for export-metadata

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.

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,15 +134,27 @@ 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}
 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,14 +164,18 @@ 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.
+
+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;
 

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

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<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 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<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();
+          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<String> 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<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();
+              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)
   {