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,7 +164,7 @@ 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.
 

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,84 @@ public void createAuditTable()
     }
   }
 
+  @Override
+  public void exportTable(
+      final String tableName,
+      final String outputPath
+  )
+  {
+    exportTableWithJdbc(tableName, outputPath);
+  }
+
+  /**
+   * Exports a table to a CSV file using generic JDBC.
+   * Binary columns are hex-encoded and booleans are written as true/false strings.
+   * Subclasses may override {@link #exportTable} with a database-specific implementation
+   * while this method remains available for testing or fallback.
+   */
+  protected void exportTableWithJdbc(
+      final String tableName,
+      final String outputPath
+  )
+  {
+    // Use a transaction so that the connection has autoCommit=false.
+    // PostgreSQL JDBC requires autoCommit=false and a positive fetch size
+    // to use cursor-based streaming instead of buffering the entire ResultSet.
+    retryTransaction(
+            (TransactionCallback<Void>) (handle, status) -> {
+              final Connection conn = handle.getConnection();
+              try (Statement stmt = conn.createStatement()) {
+                final int fetchSize = getStreamingFetchSize();
+                if (fetchSize > 0) {
+                  stmt.setFetchSize(fetchSize);
+                }
+                try (ResultSet rs = stmt.executeQuery(StringUtils.format("SELECT * FROM %s", tableName));
+                     FileOutputStream fos = new FileOutputStream(outputPath);
+                     OutputStreamWriter writer = new OutputStreamWriter(fos, StandardCharsets.UTF_8)) {
+                  final ResultSetMetaData meta = rs.getMetaData();
+                  final int columnCount = meta.getColumnCount();
+                  while (rs.next()) {
+                    for (int i = 1; i <= columnCount; i++) {
+                      if (i > 1) {
+                        writer.write(',');
+                      }
+                      final int colType = meta.getColumnType(i);
+                      if (colType == Types.BINARY || colType == Types.VARBINARY
+                          || colType == Types.LONGVARBINARY || colType == Types.BLOB
+                          || (colType == Types.OTHER && "bytea".equalsIgnoreCase(meta.getColumnTypeName(i)))) {
+                        final byte[] bytes = rs.getBytes(i);
+                        if (bytes != null) {
+                          writer.write(BaseEncoding.base16().encode(bytes));
+                        }
+                      } else if (colType == Types.BOOLEAN || colType == Types.BIT) {
+                        final boolean val = rs.getBoolean(i);
+                        if (!rs.wasNull()) {
+                          writer.write(String.valueOf(val));
+                        }
+                      } else {
+                        final String val = rs.getString(i);
+                        if (val != null) {
+                          if (val.contains(",") || val.contains("\"") || val.contains("\n") || val.contains("\r")) {
+                            writer.write('"');
+                            writer.write(StringUtils.replace(val, "\"", "\"\""));
+                            writer.write('"');
+                          } else {
+                            writer.write(val);
+                          }
+                        }
+                      }
+                    }
+                    writer.write('\n');
+                  }
+                }
+              }
+              return null;
+            },
+            QUIET_RETRIES,
+            DEFAULT_MAX_TRIES
+    );
+  }
+
   @Override
   public void deleteAllRecords(final String tableName)
   {