This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new cad5e44f155 feat: Support PostgreSQL export for export-metadata
(#19698)
cad5e44f155 is described below
commit cad5e44f155b1a8dacae4738514085595bf40204
Author: Jiang Wu <[email protected]>
AuthorDate: Fri Sep 4 04:29:16 2026 +0200
feat: Support PostgreSQL export for export-metadata (#19698)
---
docs/operations/deep-storage-migration.md | 3 +-
docs/operations/export-metadata.md | 148 +++++-
docs/operations/metadata-migration.md | 7 +-
.../storage/postgresql/PostgreSQLConnector.java | 11 +
.../druid/metadata/MetadataStorageConnector.java | 8 -
.../druid/metadata/SQLMetadataConnector.java | 49 +-
.../metadata/storage/derby/DerbyConnector.java | 25 -
.../druid/metadata/SQLMetadataConnectorTest.java | 32 ++
.../java/org/apache/druid/cli/CreateTables.java | 1 +
.../java/org/apache/druid/cli/ExportMetadata.java | 409 ++++++++-------
.../org/apache/druid/cli/MetadataCsvExporter.java | 242 +++++++++
.../org/apache/druid/cli/CreateTablesTest.java | 2 +
.../org/apache/druid/cli/ExportMetadataTest.java | 576 +++++++++++++++++++++
.../apache/druid/cli/MetadataCsvExporterTest.java | 250 +++++++++
website/.spelling | 1 +
15 files changed, 1498 insertions(+), 266 deletions(-)
diff --git a/docs/operations/deep-storage-migration.md
b/docs/operations/deep-storage-migration.md
index 733db8bd924..e72f44f5896 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 e065e42b013..4727d9d45df 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,19 @@ 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
+```
+
+If the cluster sets `druid.metadata.postgres.dbTableSchema`, pass that
property as well, since the tool looks the tables up in the configured schema
and defaults to `public`.
+
+In the example commands above:
- `lib` is the Druid lib directory
- `extensions` is the Druid extensions directory
@@ -149,54 +164,133 @@ In the example command above:
## Importing Metadata
-After running the tool, the output directory will contain
`<table-name>_raw.csv` and `<table-name>.csv` files.
+After running the tool, the output directory contains one `<table-name>.csv`
file per exported table, with any deep storage rewrites applied. Example import
commands for Derby, MySQL, and PostgreSQL are shown below. They expect
`/tmp/csv` and its contents to be accessible from the database server; for
other options, such as importing from the client filesystem, see your
database's documentation.
-The `<table-name>_raw.csv` files are intermediate files used by the tool,
containing the table data as exported by Derby without modification.
+### Before you import
-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.
+#### Adjust the column lists
+
+Every table is exported in a fixed column order, whatever the physical column
order of the source table is. For the segments table this is `id`,
`dataSource`, `created_date`, `start`, `end`, `partitioned`, `version`, `used`,
`payload`, followed by whichever of `used_status_last_updated`,
`indexing_state_fingerprint`, `upgraded_from_segment_id`, `schema_fingerprint`,
and `num_rows` the source table has, in that order.
+
+Make the segments column list in the commands below match the source table
exactly: drop the optional columns it does not have, and add
`schema_fingerprint,num_rows` at the end if it has them. Segments tables from
older Druid versions may have only the first nine columns. Adjust the
`FORCE_NULL` list in the PostgreSQL command in the same way. In the MySQL
command, each optional column needs both a user variable in the column list and
an assignment in `SET`, for example `@num_rows` with ` [...]
+
+#### Import the fingerprint tables as well
+
+`druid_segmentSchemas` and `druid_indexingStates` hold the schemas and
indexing states that the `schema_fingerprint` and `indexing_state_fingerprint`
of a segment refer to. They are exported when the source metadata store has
them, and must be imported along with the segments table, so that imported
segments do not refer to missing rows. The `id` of `druid_segmentSchemas` is a
generated identity column that nothing refers to, so it is not exported and the
target assigns new values.
+
+Both tables must exist in the target before their import commands run. Druid
creates them at startup, and so does the `metadata-init` tool described in
[Metadata migration](metadata-migration.md). Neither creates
`druid_segmentSchemas`, or the `schema_fingerprint` and `num_rows` columns of
`druid_segments`, unless `druid.centralizedDatasourceSchema.enabled` is `true`,
and `metadata-init` does not read the runtime properties of the target cluster.
So pass `-Ddruid.centralizedDatasourceSch [...]
+
+#### Fill in `used_status_last_updated`
+
+Druid creates this column as `NOT NULL` in a new segments table, but adds it
as nullable to an existing one and only sets it on the segments it marks
unused. The import therefore fails both when the source table does not have the
column and when it has the column with NULL rows. In either case, make the
column nullable, import, then fill in the missing values:
+
+```sql
+ALTER TABLE druid_segments ALTER COLUMN used_status_last_updated NULL;
+-- run the import command for your database here
+-- replace <migration-time> with the current UTC time, for example
2026-09-02T19:00:00.000Z
+UPDATE druid_segments SET used_status_last_updated = '<migration-time>' WHERE
used_status_last_updated IS NULL;
+ALTER TABLE druid_segments ALTER COLUMN used_status_last_updated NOT NULL;
+```
-Example import commands for Derby, MySQL, and PostgreSQL are shown below.
+Use the current UTC time of the migration, in ISO 8601 as above, and not
`created_date`. `created_date` is when a segment was published, while
`used_status_last_updated` is when it became unused, and the Coordinator
permanently deletes an unused segment, including from deep storage, once it has
been unused for longer than the kill task's `durationToRetain`. With
`created_date`, an old segment that was marked unused recently would be deleted
straight away.
-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 `ALTER TABLE` syntax above is Derby's. On PostgreSQL, use `ALTER COLUMN
used_status_last_updated DROP NOT NULL` and `SET NOT NULL`; on MySQL, `MODIFY
used_status_last_updated VARCHAR(255) NULL` and `MODIFY
used_status_last_updated VARCHAR(255) NOT NULL`.
+
+#### Empty fields and backslashes
+
+A NULL is exported as an unquoted empty field and an empty string as a quoted
one (`""`). Each database needs to be told how to read them:
+
+- Derby reads an unquoted empty field as NULL and a quoted one as an empty
string, so it needs no extra handling.
+- PostgreSQL `COPY` reads an empty field as an empty string, which fails for
non-string columns such as `num_rows`, so the command below lists the nullable
columns in `FORCE_NULL`. That applies to unquoted empty fields only, so empty
strings survive.
+- MySQL `LOAD DATA` also reads an empty field as an empty string, and turns it
into `0` for numeric columns such as `num_rows`, so the command below reads the
nullable columns into user variables and applies `NULLIF`. MySQL cannot tell a
quoted empty field from an unquoted one, so an empty string in such a column
becomes NULL.
+
+The exported CSV follows RFC 4180, where a backslash is an ordinary character.
MySQL `LOAD DATA` treats it as an escape character by default, which would
corrupt payloads and segment ids containing one, so the commands below turn
that off with `ESCAPED BY ''`.
### Derby
+`SYSCS_IMPORT_DATA` matches its insert-column list against the stored column
names, which are case-sensitive and must not be quoted. Derby stores the
unquoted column names of the metadata tables in uppercase, so the lists below
use uppercase names. The exception is `end`, a reserved word that Druid creates
as a quoted lowercase identifier: `"END"` and `END` both fail with `XIE08 There
is no column named`.
+
+The `ID` of `DRUID_SUPERVISORS` and `DRUID_SEGMENTSCHEMAS` is `GENERATED
ALWAYS AS IDENTITY`, which Derby refuses to insert a value into, so it is left
out of the lists below and Derby assigns new ids. The segment schemas CSV has
no `id` and imports directly; the supervisors CSV starts with its `ID` and
needs the staging table shown after the commands.
+
```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_DATA
(null,'DRUID_RULES','ID,DATASOURCE,VERSION,PAYLOAD',null,'/tmp/csv/druid_rules.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_CONFIG','NAME,PAYLOAD',null,'/tmp/csv/druid_config.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_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_DATASOURCE','/tmp/csv/druid_dataSource.csv',',','"',null,0);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_SEGMENTSCHEMAS','FINGERPRINT,CREATED_DATE,DATASOURCE,PAYLOAD,USED,USED_STATUS_LAST_UPDATED,VERSION',null,'/tmp/csv/druid_segmentSchemas.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_INDEXINGSTATES','FINGERPRINT,CREATED_DATE,DATASOURCE,PAYLOAD,USED,PENDING,USED_STATUS_LAST_UPDATED',null,'/tmp/csv/druid_indexingStates.csv',',','"',null,0);
+```
+
+Import the supervisors through a staging table, which drops the exported `ID`
and lets Derby generate new ones. Druid reads the latest version of a
supervisor spec by `ID`, so insert the rows ordered by the exported id:
+
+```sql
+CREATE TABLE DRUID_SUPERVISORS_IMPORT (ID BIGINT, SPEC_ID VARCHAR(255),
CREATED_DATE VARCHAR(255), PAYLOAD BLOB);
+CALL SYSCS_UTIL.SYSCS_IMPORT_DATA
(null,'DRUID_SUPERVISORS_IMPORT','ID,SPEC_ID,CREATED_DATE,PAYLOAD',null,'/tmp/csv/druid_supervisors.csv',',','"',null,0);
+INSERT INTO DRUID_SUPERVISORS (SPEC_ID, CREATED_DATE, PAYLOAD) SELECT SPEC_ID,
CREATED_DATE, PAYLOAD FROM DRUID_SUPERVISORS_IMPORT ORDER BY ID;
+DROP TABLE DRUID_SUPERVISORS_IMPORT;
```
### 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,''); SHO [...]
-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
+LOAD DATA INFILE '/tmp/csv/druid_segmentSchemas.csv' INTO TABLE
druid_segmentSchemas FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'
ESCAPED BY ''
(fingerprint,created_date,datasource,payload,used,used_status_last_updated,version);
SHOW WARNINGS;
-```sql
-COPY
druid_segments(id,dataSource,created_date,start,"end",partitioned,version,used,payload)
FROM '/tmp/csv/druid_segments.csv' DELIMITER ',' CSV;
+LOAD DATA INFILE '/tmp/csv/druid_indexingStates.csv' INTO TABLE
druid_indexingStates FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'
ESCAPED BY ''
(fingerprint,created_date,dataSource,payload,used,pending,used_status_last_updated);
SHOW WARNINGS;
+```
-COPY druid_rules(id,dataSource,version,payload) FROM
'/tmp/csv/druid_rules.csv' DELIMITER ',' CSV;
+The supervisors command keeps the exported `id` values; InnoDB advances its
auto-increment counter past them, so no reseeding is needed.
-COPY druid_config(name,payload) FROM '/tmp/csv/druid_config.csv' DELIMITER ','
CSV;
+### PostgreSQL
+
+The payload columns are `BYTEA` in PostgreSQL, and `COPY` reads a `BYTEA`
field with the `bytea` input syntax, where a backslash starts an escape
sequence. Exported payloads are JSON text that commonly contains backslashes,
such as escaped quotes or Windows deep storage paths, so loading them straight
into a `BYTEA` column either fails or silently changes the payload.
-COPY
druid_dataSource(dataSource,created_date,commit_metadata_payload,commit_metadata_sha1)
FROM '/tmp/csv/druid_dataSource.csv' DELIMITER ',' CSV;
+Run the export with `--use-hex-blobs` (`-x`) to keep the payloads hex-encoded,
load the CSV files into staging tables whose payload columns are `TEXT`, and
decode with `decode(payload, 'hex')`:
-COPY druid_supervisors(id,spec_id,created_date,payload) FROM
'/tmp/csv/druid_supervisors.csv' DELIMITER ',' CSV;
+```sql
+CREATE TEMP TABLE druid_segments_import(id text,dataSource text,created_date
text,start text,"end" text,partitioned boolean,version text,used
boolean,payload text,used_status_last_updated text,indexing_state_fingerprint
text,upgraded_from_segment_id text);
+COPY druid_segments_import FROM '/tmp/csv/druid_segments.csv' WITH (FORMAT
csv, FORCE_NULL
(used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id));
+INSERT INTO
druid_segments(id,dataSource,created_date,start,"end",partitioned,version,used,payload,used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id)
SELECT
id,dataSource,created_date,start,"end",partitioned,version,used,decode(payload,'hex'),used_status_last_updated,indexing_state_fingerprint,upgraded_from_segment_id
FROM druid_segments_import;
+
+CREATE TEMP TABLE druid_rules_import(id text,dataSource text,version
text,payload text);
+COPY druid_rules_import FROM '/tmp/csv/druid_rules.csv' WITH (FORMAT csv);
+INSERT INTO druid_rules(id,dataSource,version,payload) SELECT
id,dataSource,version,decode(payload,'hex') FROM druid_rules_import;
+
+CREATE TEMP TABLE druid_config_import(name text,payload text);
+COPY druid_config_import FROM '/tmp/csv/druid_config.csv' WITH (FORMAT csv);
+INSERT INTO druid_config(name,payload) SELECT name,decode(payload,'hex') FROM
druid_config_import;
+
+CREATE TEMP TABLE druid_dataSource_import(dataSource text,created_date
text,commit_metadata_payload text,commit_metadata_sha1 text);
+COPY druid_dataSource_import FROM '/tmp/csv/druid_dataSource.csv' WITH (FORMAT
csv);
+INSERT INTO
druid_dataSource(dataSource,created_date,commit_metadata_payload,commit_metadata_sha1)
SELECT
dataSource,created_date,decode(commit_metadata_payload,'hex'),commit_metadata_sha1
FROM druid_dataSource_import;
+
+CREATE TEMP TABLE druid_supervisors_import(id text,spec_id text,created_date
text,payload text);
+COPY druid_supervisors_import FROM '/tmp/csv/druid_supervisors.csv' WITH
(FORMAT csv);
+INSERT INTO druid_supervisors(id,spec_id,created_date,payload) SELECT
id::bigint,spec_id,created_date,decode(payload,'hex') FROM
druid_supervisors_import;
+-- druid_supervisors.id is a BIGSERIAL, whose sequence must be advanced past
the imported ids
+SELECT setval(pg_get_serial_sequence('druid_supervisors','id'), (SELECT
MAX(id) FROM druid_supervisors));
+
+CREATE TEMP TABLE druid_segmentSchemas_import(fingerprint text,created_date
text,datasource text,payload text,used boolean,used_status_last_updated
text,version integer);
+COPY druid_segmentSchemas_import FROM '/tmp/csv/druid_segmentSchemas.csv' WITH
(FORMAT csv);
+INSERT INTO
druid_segmentSchemas(fingerprint,created_date,datasource,payload,used,used_status_last_updated,version)
SELECT
fingerprint,created_date,datasource,decode(payload,'hex'),used,used_status_last_updated,version
FROM druid_segmentSchemas_import;
+
+CREATE TEMP TABLE druid_indexingStates_import(fingerprint text,created_date
text,dataSource text,payload text,used boolean,pending
boolean,used_status_last_updated text);
+COPY druid_indexingStates_import FROM '/tmp/csv/druid_indexingStates.csv' WITH
(FORMAT csv);
+INSERT INTO
druid_indexingStates(fingerprint,created_date,dataSource,payload,used,pending,used_status_last_updated)
SELECT
fingerprint,created_date,dataSource,decode(payload,'hex'),used,pending,used_status_last_updated
FROM druid_indexingStates_import;
```
+
+The staging tables declare the columns the export writes; adjust the segments
columns as described above if the source table has a different set of optional
ones. A staging column is `text` wherever the exported value needs a cast or a
decode, such as the `id` of `druid_supervisors`, which is a `BIGSERIAL` in the
target table.
diff --git a/docs/operations/metadata-migration.md
b/docs/operations/metadata-migration.md
index ea3596784ad..ab616adba12 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
@@ -70,6 +71,8 @@ In the example commands below:
- The `--user` parameter corresponds to the value of
`druid.metadata.storage.connector.user`.
- The `--password` parameter corresponds to the value of
`druid.metadata.storage.connector.password`.
+The commands do not read the runtime properties of the cluster, so any
property which affects the tables must be passed on the command line. In
particular, the `druid_segmentSchemas` table and the `schema_fingerprint` and
`num_rows` columns of `druid_segments` are only created with
`-Ddruid.centralizedDatasourceSchema.enabled=true`. Pass it if the cluster uses
centralized datasource schema, or if it is being populated from a metadata
store which has those tables and columns: disabling th [...]
+
#### MySQL
```bash
diff --git
a/extensions-core/postgresql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/postgresql/PostgreSQLConnector.java
b/extensions-core/postgresql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/postgresql/PostgreSQLConnector.java
index 48e0bd4e8c3..0a5af54bbfe 100644
---
a/extensions-core/postgresql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/postgresql/PostgreSQLConnector.java
+++
b/extensions-core/postgresql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/postgresql/PostgreSQLConnector.java
@@ -38,6 +38,7 @@ import org.skife.jdbi.v2.exceptions.CallbackFailedException;
import org.skife.jdbi.v2.tweak.HandleCallback;
import org.skife.jdbi.v2.util.StringMapper;
+import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.util.List;
@@ -177,6 +178,16 @@ public class PostgreSQLConnector extends
SQLMetadataConnector
.isEmpty();
}
+ /**
+ * Scopes metadata lookups to the same schema as {@link #tableExists}, which
may differ from the
+ * schema of the connection when the {@code search_path} resolves
unqualified names elsewhere.
+ */
+ @Override
+ public String getMetadataTableSchema(final Connection connection)
+ {
+ return dbTableSchema;
+ }
+
@Override
public Void insertOrUpdate(
final String tableName,
diff --git
a/processing/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java
b/processing/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java
index f40b36efa17..79076014688 100644
---
a/processing/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java
+++
b/processing/src/main/java/org/apache/druid/metadata/MetadataStorageConnector.java
@@ -63,14 +63,6 @@ public interface MetadataStorageConnector
throw new UnsupportedOperationException("compareAndSwap is not
implemented.");
}
- default void exportTable(
- String tableName,
- String outputPath
- )
- {
- throw new UnsupportedOperationException("exportTable is not implemented.");
- }
-
void createDataSourceTable();
void createPendingSegmentsTable();
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 f76f6feb907..aeb38dc25d5 100644
--- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java
+++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java
@@ -74,8 +74,8 @@ public abstract class SQLMetadataConnector implements
MetadataStorageConnector
private static final String PAYLOAD_TYPE = "BLOB";
private static final String COLLATION = "";
- static final int QUIET_RETRIES = 2;
- static final int DEFAULT_MAX_TRIES = 3;
+ public static final int QUIET_RETRIES = 2;
+ public static final int DEFAULT_MAX_TRIES = 3;
private final Supplier<MetadataStorageConnectorConfig> config;
private final Supplier<MetadataStorageTablesConfig> tablesConfigSupplier;
@@ -1055,6 +1055,51 @@ public abstract class SQLMetadataConnector implements
MetadataStorageConnector
}
}
+ /**
+ * Returns the columns of the given table, in the order reported by the
database, or an empty list if the table
+ * does not exist. Throws if the database metadata cannot be read, so that a
caller which acts on a table being
+ * absent does not mistake a failed lookup for an absent table.
+ *
+ * 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. The schema is a search pattern
too, so the returned schema is compared
+ * to it as well, in case the configured schema contains a '_' or '%' which
would otherwise match other schemas.
+ */
+ public List<String> getTableColumns(final String tableName)
+ {
+ return getDBI().withHandle(handle -> {
+ final List<String> columns = new ArrayList<>();
+ if (tableExists(handle, tableName)) {
+ final Connection conn = handle.getConnection();
+ final String schema = getMetadataTableSchema(conn);
+ try (ResultSet rs = conn.getMetaData().getColumns(null, schema, null,
null)) {
+ while (rs.next()) {
+ if (tableName.equalsIgnoreCase(rs.getString("TABLE_NAME"))
+ && (schema == null ||
schema.equals(rs.getString("TABLE_SCHEM")))) {
+ columns.add(rs.getString("COLUMN_NAME"));
+ }
+ }
+ }
+ }
+ 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
+ public String getMetadataTableSchema(final Connection connection) throws
SQLException
+ {
+ return connection.getSchema();
+ }
+
@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 cb4815d2313..4fb322525c7 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
@@ -128,31 +128,6 @@ public class DerbyConnector extends SQLMetadataConnector
return String.format(Locale.ENGLISH, "FETCH NEXT %d ROWS ONLY", limit);
}
- @Override
- public void exportTable(
- String tableName,
- String outputPath
- )
- {
- retryWithHandle(
- new HandleCallback<Void>()
- {
- @Override
- public Void withHandle(Handle handle)
- {
- handle.createStatement(
- StringUtils.format(
- "CALL SYSCS_UTIL.SYSCS_EXPORT_TABLE (null, '%s', '%s',
null, null, null)",
- tableName,
- outputPath
- )
- ).execute();
- return null;
- }
- }
- );
- }
-
/**
* Get the ResultSet for indexInfo for given table
*
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 4a4f3f236a6..9ef9ba5d7c2 100644
---
a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java
+++
b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java
@@ -21,6 +21,7 @@ package org.apache.druid.metadata;
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;
@@ -469,6 +470,37 @@ public class SQLMetadataConnectorTest
);
}
+ @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;
+ }
+ );
+
+ Assertions.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
+ Assertions.assertEquals(
+ ImmutableList.of("ID", "USED"),
+ connector.getTableColumns(StringUtils.toLowerCase(tableName))
+ );
+ Assertions.assertEquals(ImmutableList.of(),
connector.getTableColumns("NON_EXISTENT_TABLE"));
+
+ dropTable(tableName);
+ }
+
static class TestSQLMetadataConnector extends SQLMetadataConnector
{
public TestSQLMetadataConnector(
diff --git a/services/src/main/java/org/apache/druid/cli/CreateTables.java
b/services/src/main/java/org/apache/druid/cli/CreateTables.java
index e332448a098..68399a3223f 100644
--- a/services/src/main/java/org/apache/druid/cli/CreateTables.java
+++ b/services/src/main/java/org/apache/druid/cli/CreateTables.java
@@ -125,6 +125,7 @@ public class CreateTables extends GuiceRunnable
dbConnector.createPendingSegmentsTable();
dbConnector.createSegmentSchemasTable();
dbConnector.createSegmentTable();
+ dbConnector.createIndexingStatesTable();
dbConnector.createUpgradeSegmentsTable();
dbConnector.createRulesTable();
dbConnector.createConfigTable();
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 f6c6e510f64..c5b548da14c 100644
--- a/services/src/main/java/org/apache/druid/cli/ExportMetadata.java
+++ b/services/src/main/java/org/apache/druid/cli/ExportMetadata.java
@@ -26,16 +26,18 @@ import com.github.rvesse.airline.annotations.Option;
import com.github.rvesse.airline.annotations.restrictions.Required;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Module;
-import com.opencsv.CSVParser;
+import org.apache.druid.cli.MetadataCsvExporter.ColumnKind;
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;
@@ -48,20 +50,20 @@ import
org.apache.druid.timeline.DataSegment.PruneSpecsHolder;
import javax.annotation.Nullable;
import javax.xml.bind.DatatypeConverter;
-import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
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 +122,82 @@ 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);
+ /**
+ * How one metadata table is exported: the canonical order in which its
columns are written, matching the import
+ * commands documented in {@code docs/operations/export-metadata.md}, the
columns which are not exported (in lower
+ * case), and whether the table may be absent from the metadata store, in
which case it is skipped instead of
+ * failing the export.
+ */
+ static final class TableSpec
+ {
+ final List<String> columnOrder;
+ final Set<String> excludedColumns;
+ final boolean optional;
+
+ TableSpec(final List<String> columnOrder)
+ {
+ this(columnOrder, ImmutableSet.of(), false);
+ }
+
+ TableSpec(final List<String> columnOrder, final Set<String>
excludedColumns, final boolean optional)
+ {
+ this.columnOrder = columnOrder;
+ this.excludedColumns = excludedColumns;
+ this.optional = optional;
+ }
+ }
+
+ private static final TableSpec DATASOURCE = new TableSpec(
+ ImmutableList.of("dataSource", "created_date",
"commit_metadata_payload", "commit_metadata_sha1")
+ );
+
+ private static final TableSpec RULES = new TableSpec(
+ ImmutableList.of("id", "dataSource", "version", "payload")
+ );
+
+ private static final TableSpec CONFIG = new TableSpec(
+ ImmutableList.of("name", "payload")
+ );
+
+ private static final TableSpec SUPERVISORS = new TableSpec(
+ ImmutableList.of("id", "spec_id", "created_date", "payload")
+ );
+
+ static final TableSpec SEGMENTS = new TableSpec(
+ 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"
+ )
+ );
+
+ /**
+ * The table holding the schemas that the {@code schema_fingerprint} of a
segment refers to. It only exists if
+ * centralized datasource schema is enabled, and its {@code id} is a
generated identity column which nothing
+ * refers to, so it is not exported: importing a value into such a column is
rejected outright by Derby and needs
+ * database-specific handling elsewhere, while letting the target database
generate the ids works everywhere.
+ */
+ static final TableSpec SEGMENT_SCHEMAS = new TableSpec(
+ ImmutableList.of(
+ "fingerprint", "created_date", "datasource", "payload", "used",
"used_status_last_updated", "version"
+ ),
+ ImmutableSet.of("id"),
+ true
+ );
+
+ /**
+ * The table holding the indexing states that the {@code
indexing_state_fingerprint} of a segment refers to. It
+ * does not exist in metadata stores written by older Druid versions.
+ */
+ static final TableSpec INDEXING_STATES = new TableSpec(
+ ImmutableList.of(
+ "fingerprint", "created_date", "dataSource", "payload", "used",
"pending", "used_status_last_updated"
+ ),
+ ImmutableSet.of(),
+ true
+ );
- private static final CSVParser PARSER = new CSVParser();
+ private static final Logger log = new Logger(ExportMetadata.class);
private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper();
@@ -184,10 +259,7 @@ public class ExportMetadata extends GuiceRunnable
@Override
public void run()
{
- InjectableValues.Std injectableValues = new InjectableValues.Std();
- injectableValues.addValue(ObjectMapper.class, JSON_MAPPER);
- injectableValues.addValue(PruneSpecsHolder.class,
PruneSpecsHolder.DEFAULT);
- JSON_MAPPER.setInjectableValues(injectableValues);
+ configureJsonMapper();
if (hadoopStorageDirectory != null && newLocalPath != null) {
throw new IllegalArgumentException(
@@ -206,212 +278,157 @@ public class ExportMetadata extends GuiceRunnable
}
final Injector injector = makeInjector();
- SQLMetadataConnector dbConnector =
injector.getInstance(SQLMetadataConnector.class);
- MetadataStorageTablesConfig metadataStorageTablesConfig =
injector.getInstance(MetadataStorageTablesConfig.class);
-
- // We export a raw CSV first, and then apply some conversions for easier
imports:
- // Boolean strings are rewritten as 1 and 0
- // hexadecimal BLOB columns are rewritten with
rewriteHexPayloadAsEscapedJson()
- log.info("Exporting datasource table: " +
metadataStorageTablesConfig.getDataSourceTable());
- exportTable(dbConnector, metadataStorageTablesConfig.getDataSourceTable(),
true);
- rewriteDatasourceExport(metadataStorageTablesConfig.getDataSourceTable());
-
- log.info("Exporting segments table: " +
metadataStorageTablesConfig.getSegmentsTable());
- exportTable(dbConnector, metadataStorageTablesConfig.getSegmentsTable(),
true);
- rewriteSegmentsExport(metadataStorageTablesConfig.getSegmentsTable());
-
- log.info("Exporting rules table: " +
metadataStorageTablesConfig.getRulesTable());
- exportTable(dbConnector, metadataStorageTablesConfig.getRulesTable(),
true);
- rewriteRulesExport(metadataStorageTablesConfig.getRulesTable());
-
- log.info("Exporting config table: " +
metadataStorageTablesConfig.getConfigTable());
- exportTable(dbConnector, metadataStorageTablesConfig.getConfigTable(),
true);
- rewriteConfigExport(metadataStorageTablesConfig.getConfigTable());
-
- log.info("Exporting supervisor table: " +
metadataStorageTablesConfig.getSupervisorTable());
- exportTable(dbConnector, metadataStorageTablesConfig.getSupervisorTable(),
true);
- rewriteSupervisorExport(metadataStorageTablesConfig.getSupervisorTable());
+ final SQLMetadataConnector dbConnector =
injector.getInstance(SQLMetadataConnector.class);
+ final MetadataStorageTablesConfig tablesConfig =
injector.getInstance(MetadataStorageTablesConfig.class);
+ final MetadataCsvExporter exporter = new MetadataCsvExporter(dbConnector);
+
+ exportTable(exporter, dbConnector, tablesConfig.getDataSourceTable(),
DATASOURCE, this::convertValue);
+ exportTable(exporter, dbConnector, tablesConfig.getSegmentsTable(),
SEGMENTS, this::convertSegmentValue);
+ exportTable(exporter, dbConnector, tablesConfig.getRulesTable(), RULES,
this::convertValue);
+ exportTable(exporter, dbConnector, tablesConfig.getConfigTable(), CONFIG,
this::convertValue);
+ exportTable(exporter, dbConnector, tablesConfig.getSupervisorTable(),
SUPERVISORS, this::convertValue);
+ // The tables the fingerprints of a segment refer to. Without them, an
imported segment refers to a schema and an
+ // indexing state which the target metadata store does not have.
+ exportTable(exporter, dbConnector, tablesConfig.getSegmentSchemasTable(),
SEGMENT_SCHEMAS, this::convertValue);
+ exportTable(exporter, dbConnector, tablesConfig.getIndexingStatesTable(),
INDEXING_STATES, this::convertValue);
}
- private void exportTable(
- SQLMetadataConnector dbConnector,
- String tableName,
- boolean withRawFilename
- )
+ static void configureJsonMapper()
{
- String pathFormatString;
- if (withRawFilename) {
- pathFormatString = "%s/%s_raw.csv";
- } else {
- pathFormatString = "%s/%s.csv";
- }
- dbConnector.exportTable(
- StringUtils.toUpperCase(tableName),
- StringUtils.format(pathFormatString, outputPath, tableName)
- );
+ final InjectableValues.Std injectableValues = new InjectableValues.Std();
+ injectableValues.addValue(ObjectMapper.class, JSON_MAPPER);
+ injectableValues.addValue(PruneSpecsHolder.class,
PruneSpecsHolder.DEFAULT);
+ JSON_MAPPER.setInjectableValues(injectableValues);
}
- private void rewriteDatasourceExport(
- String datasourceTableName
+ /**
+ * Exports one metadata table to {@code <outputPath>/<tableName>.csv},
writing the columns of the source table in
+ * the canonical order of the given spec.
+ */
+ void exportTable(
+ final MetadataCsvExporter exporter,
+ final SQLMetadataConnector dbConnector,
+ final String tableName,
+ final TableSpec spec,
+ final MetadataCsvExporter.ValueConverter converter
)
{
- 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)
- );
- 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
- + rewriteHexPayloadAsEscapedJson(parsed[2]) + ","
//commit_metadata_payload
- + parsed[3] //commit_metadata_sha1
- + "\n";
- writer.write(newLine);
-
+ // Derby folds unquoted identifiers to upper case, so its tables are named
in upper case, while PostgreSQL folds
+ // them to lower case and uses the table names as configured.
+ final String sourceTableName = isDerby() ?
StringUtils.toUpperCase(tableName) : tableName;
+ final List<String> columns =
+ orderColumns(spec.columnOrder,
dbConnector.getTableColumns(sourceTableName), spec.excludedColumns);
+ // An empty column list means the table does not exist: a failed lookup
throws instead, so that a table which
+ // cannot be read is never mistaken for one the metadata store does not
have.
+ if (columns.isEmpty()) {
+ if (spec.optional) {
+ // Remove the file of an earlier export into the same directory, which
would otherwise be imported as this
+ // table's data.
+ deleteOutputFile(tableName);
+ log.info("Skipping table[%s], which this metadata store does not
have.", tableName);
+ return;
}
+ throw new ISE("Table[%s] does not exist in this metadata store.",
sourceTableName);
}
- catch (IOException ioex) {
- throw new RuntimeException(ioex);
- }
+
+ log.info("Exporting table[%s].", tableName);
+ exporter.exportTable(sourceTableName, columns, makeOutputFile(tableName),
converter);
}
- private void rewriteRulesExport(
- String rulesTableName
- )
+ private Path makeOutputFile(final String tableName)
{
- 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)
- );
- 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
- + rewriteHexPayloadAsEscapedJson(parsed[3]) //payload
- + "\n";
- writer.write(newLine);
+ return Paths.get(outputPath, StringUtils.format("%s.csv", tableName));
+ }
- }
+ private void deleteOutputFile(final String tableName)
+ {
+ final Path outputFile = makeOutputFile(tableName);
+ try {
+ Files.deleteIfExists(outputFile);
}
- catch (IOException ioex) {
- throw new RuntimeException(ioex);
+ catch (IOException e) {
+ throw new ISE(e, "Could not delete stale file[%s] of skipped
table[%s].", outputFile, tableName);
}
}
- private void rewriteConfigExport(
- String configTableName
+ /**
+ * Orders the given actual column names of a table by the given canonical
order, ignoring case, skipping canonical
+ * columns which the table does not have and appending any unknown columns
at the end in their original order.
+ * Columns in {@code excludedColumns} are left out. Exporting this explicit
column list keeps the output
+ * independent of the physical column order, which depends on the order in
which {@code ALTER TABLE} added the
+ * newer columns.
+ */
+ static List<String> orderColumns(
+ final List<String> columnOrder,
+ final List<String> actualColumns,
+ final Set<String> excludedColumns
)
{
- 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)
- );
- 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
- + rewriteHexPayloadAsEscapedJson(parsed[1]) //payload
- + "\n";
- writer.write(newLine);
-
+ final Map<String, String> remaining = new LinkedHashMap<>();
+ for (String column : actualColumns) {
+ final String lowerCaseColumn = StringUtils.toLowerCase(column);
+ if (!excludedColumns.contains(lowerCaseColumn)) {
+ remaining.put(lowerCaseColumn, column);
}
}
- catch (IOException ioex) {
- throw new RuntimeException(ioex);
+
+ final List<String> ordered = new ArrayList<>(actualColumns.size());
+ for (String column : columnOrder) {
+ final String actualColumn =
remaining.remove(StringUtils.toLowerCase(column));
+ if (actualColumn != null) {
+ ordered.add(actualColumn);
+ }
}
+ ordered.addAll(remaining.values());
+ return ordered;
}
- private void rewriteSupervisorExport(
- String supervisorTableName
- )
+ private boolean isDerby()
{
- 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)
- );
- 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
- + rewriteHexPayloadAsEscapedJson(parsed[3]) //payload
- + "\n";
- writer.write(newLine);
+ return connectURI != null && connectURI.startsWith("jdbc:derby");
+ }
- }
+ /**
+ * Converts one exported value: BLOB payloads are decoded as JSON unless
{@code --use-hex-blobs} was given, and
+ * booleans are written as 1 and 0 unless {@code --booleans-as-strings} was
given. A NULL stays a NULL.
+ */
+ @Nullable
+ String convertValue(final ColumnKind kind, @Nullable final String value)
+ {
+ if (value == null) {
+ return null;
}
- catch (IOException ioex) {
- throw new RuntimeException(ioex);
+ switch (kind) {
+ case BINARY:
+ return convertPayload(value);
+ case BOOLEAN:
+ return convertBooleanString(value);
+ default:
+ return value;
}
}
-
- private void rewriteSegmentsExport(
- String segmentsTableName
- )
+ /**
+ * Converts one exported value of the segments table, rewriting the deep
storage location of the segment payload
+ * if one of {@code --s3bucket}, {@code --hadoopStorageDirectory} or {@code
--newLocalPath} was given, and
+ * otherwise behaving like {@link #convertValue}.
+ */
+ @Nullable
+ String convertSegmentValue(final ColumnKind kind, @Nullable final String
value) throws IOException
{
- 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)
- );
- OutputStreamWriter writer = new OutputStreamWriter(new
FileOutputStream(outFile), StandardCharsets.UTF_8)
- ) {
- String line;
- while ((line = reader.readLine()) != null) {
- String[] parsed = PARSER.parseLine(line);
- 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(convertBooleanString(parsed[5])).append(",");
//partitioned
- newLineBuilder.append(parsed[6]).append(","); //version
- newLineBuilder.append(convertBooleanString(parsed[7])).append(",");
//used
-
- if (s3Bucket != null || hadoopStorageDirectory != null || newLocalPath
!= null) {
- newLineBuilder.append(makePayloadWithConvertedLoadSpec(parsed[8]));
- } else {
- newLineBuilder.append(rewriteHexPayloadAsEscapedJson(parsed[8]));
//payload
- }
- newLineBuilder.append("\n");
- writer.write(newLineBuilder.toString());
-
- }
- }
- catch (IOException ioex) {
- throw new RuntimeException(ioex);
+ if (kind == ColumnKind.BINARY && value != null &&
isDeepStorageMigration()) {
+ return makePayloadWithConvertedLoadSpec(value);
}
+ return convertValue(kind, value);
+ }
+
+ private boolean isDeepStorageMigration()
+ {
+ return s3Bucket != null || hadoopStorageDirectory != null || newLocalPath
!= null;
}
/**
- * Returns a new load spec in escaped JSON form, with the new deep storage
location if configured.
+ * Returns the segment payload in JSON form, with the new deep storage
location if configured.
*/
private String makePayloadWithConvertedLoadSpec(
String payload
@@ -438,28 +455,23 @@ public class ExportMetadata extends GuiceRunnable
if (useHexBlobs) {
return DatatypeConverter.printHexBinary(StringUtils.toUtf8(serialized));
} else {
- return escapeJSONForCSV(serialized);
+ return serialized;
}
}
/**
- * Derby's export tool writes BLOB columns as a hexadecimal string:
- * https://db.apache.org/derby/docs/10.9/adminguide/cadminimportlobs.html
- *
- * Decodes the hex string and escapes the decoded JSON.
+ * Converts a BLOB value, which the exporter reads as a hexadecimal string,
to the JSON it holds, unless hex blobs
+ * were requested.
*/
- private String rewriteHexPayloadAsEscapedJson(
- String payload
- )
+ private String convertPayload(final String payload)
{
if (useHexBlobs) {
return payload;
}
- String json =
StringUtils.fromUtf8(DatatypeConverter.parseHexBinary(payload));
- return escapeJSONForCSV(json);
+ return StringUtils.fromUtf8(DatatypeConverter.parseHexBinary(payload));
}
- private String convertBooleanString(String booleanString)
+ private String convertBooleanString(final String booleanString)
{
if (booleansAsStrings) {
return booleanString;
@@ -468,11 +480,6 @@ public class ExportMetadata extends GuiceRunnable
}
}
- private String escapeJSONForCSV(String json)
- {
- return "\"" + StringUtils.replace(json, "\"", "\"\"") + "\"";
- }
-
private Map<String, Object> makeS3LoadSpec(
String segmentPath
)
diff --git
a/services/src/main/java/org/apache/druid/cli/MetadataCsvExporter.java
b/services/src/main/java/org/apache/druid/cli/MetadataCsvExporter.java
new file mode 100644
index 00000000000..26b5d09a7dd
--- /dev/null
+++ b/services/src/main/java/org/apache/druid/cli/MetadataCsvExporter.java
@@ -0,0 +1,242 @@
+/*
+ * 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.io.BaseEncoding;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.skife.jdbi.v2.TransactionCallback;
+
+import javax.annotation.Nullable;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Exports metadata tables to CSV files, reading the rows from any database
through the JDBC connection of a
+ * {@link SQLMetadataConnector}.
+ *
+ * A table is read and written in a single streaming pass: every row is read
from the {@link ResultSet}, handed to a
+ * {@link ValueConverter} which applies the conversions the {@code
export-metadata} tool needs, and written as one CSV
+ * record. The connector only provides the connection, the schema of the Druid
tables and the fetch size that makes
+ * the database stream its results; all CSV concerns live here.
+ *
+ * The output follows RFC 4180: a field containing a comma, a double quote or
a line break is wrapped in double
+ * quotes, with inner double quotes doubled, and a backslash is an ordinary
character. A field which begins or ends
+ * with whitespace is quoted as well, which RFC 4180 does not require but
Derby needs in order to preserve it. A
+ * NULL is written as an unquoted empty field while an empty string is written
as a quoted empty field, so that the
+ * two remain distinguishable by the import commands documented in {@code
docs/operations/export-metadata.md}.
+ */
+public class MetadataCsvExporter
+{
+ /**
+ * The kind of a column, as far as the CSV representation of its values is
concerned.
+ */
+ public enum ColumnKind
+ {
+ /** A binary column, whose values are passed to the converter as an
upper-case hexadecimal string. */
+ BINARY,
+ /** A boolean column, whose values are passed to the converter as {@code
"true"} or {@code "false"}. */
+ BOOLEAN,
+ /** Any other column, whose values are passed to the converter as returned
by {@link ResultSet#getString}. */
+ OTHER
+ }
+
+ /**
+ * Converts one value read from the database into the value to write to the
CSV.
+ *
+ * The value is null for a NULL and is never escaped, so that a converter
does not have to know about the CSV
+ * format: escaping is applied to whatever it returns.
+ */
+ @FunctionalInterface
+ public interface ValueConverter
+ {
+ @Nullable
+ String convert(ColumnKind kind, @Nullable String value) throws Exception;
+ }
+
+ private final SQLMetadataConnector connector;
+
+ public MetadataCsvExporter(final SQLMetadataConnector connector)
+ {
+ this.connector = connector;
+ }
+
+ /**
+ * Exports the given columns of the given table to a CSV file, one record
per row, in the order the columns are
+ * given, applying {@code converter} to every value.
+ */
+ public void exportTable(
+ final String tableName,
+ final List<String> columns,
+ final Path outputPath,
+ final ValueConverter converter
+ )
+ {
+ // 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.
+ connector.retryTransaction(
+ (TransactionCallback<Void>) (handle, status) -> {
+ final Connection conn = handle.getConnection();
+ 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(connector.getStreamingFetchSize());
+ try (
+ ResultSet rs = stmt.executeQuery(makeSelectStatement(conn,
tableName, columns));
+ Writer writer = new OutputStreamWriter(
+ Files.newOutputStream(outputPath),
+ StandardCharsets.UTF_8
+ )
+ ) {
+ final ColumnKind[] kinds = readColumnKinds(rs.getMetaData());
+ while (rs.next()) {
+ writeRecord(writer, rs, kinds, converter);
+ }
+ }
+ }
+ return null;
+ },
+ SQLMetadataConnector.QUIET_RETRIES,
+ SQLMetadataConnector.DEFAULT_MAX_TRIES
+ );
+ }
+
+ /**
+ * Builds the select statement for an export, qualifying 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 and the columns are quoted, since they are the names as stored
in the database, so that reserved
+ * words such as {@code end} work. The table name is left unquoted, so that
it is folded by the database in the
+ * same way as in every other Druid statement.
+ */
+ private String makeSelectStatement(
+ final Connection conn,
+ final String tableName,
+ final List<String> columns
+ ) throws SQLException
+ {
+ final String quote = conn.getMetaData().getIdentifierQuoteString();
+ final String selectList = columns.stream()
+ .map(column -> quoteIdentifier(quote,
column))
+ .collect(Collectors.joining(","));
+ final String schema = connector.getMetadataTableSchema(conn);
+ final String qualifiedTableName =
+ schema == null ? tableName : quoteIdentifier(quote, schema) + "." +
tableName;
+ return StringUtils.format("SELECT %s FROM %s", selectList,
qualifiedTableName);
+ }
+
+ /**
+ * 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 java.sql.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;
+ }
+
+ private static ColumnKind[] readColumnKinds(final ResultSetMetaData meta)
throws SQLException
+ {
+ final ColumnKind[] kinds = new ColumnKind[meta.getColumnCount()];
+ for (int i = 0; i < kinds.length; i++) {
+ final int type = meta.getColumnType(i + 1);
+ if (type == Types.BINARY || type == Types.VARBINARY || type ==
Types.LONGVARBINARY || type == Types.BLOB
+ // PostgreSQL reports BYTEA as OTHER
+ || (type == Types.OTHER &&
"bytea".equalsIgnoreCase(meta.getColumnTypeName(i + 1)))) {
+ kinds[i] = ColumnKind.BINARY;
+ } else if (type == Types.BOOLEAN || type == Types.BIT) {
+ kinds[i] = ColumnKind.BOOLEAN;
+ } else {
+ kinds[i] = ColumnKind.OTHER;
+ }
+ }
+ return kinds;
+ }
+
+ @Nullable
+ private static String readValue(final ResultSet rs, final int column, final
ColumnKind kind) throws SQLException
+ {
+ switch (kind) {
+ case BINARY:
+ final byte[] bytes = rs.getBytes(column);
+ return bytes == null ? null : BaseEncoding.base16().encode(bytes);
+ case BOOLEAN:
+ final boolean value = rs.getBoolean(column);
+ return rs.wasNull() ? null : String.valueOf(value);
+ default:
+ return rs.getString(column);
+ }
+ }
+
+ /**
+ * Converts and writes the current row as one CSV record, leaving a NULL as
null on the way to the converter so
+ * that it stays distinguishable from an empty string.
+ */
+ private static void writeRecord(
+ final Writer writer,
+ final ResultSet rs,
+ final ColumnKind[] kinds,
+ final ValueConverter converter
+ ) throws Exception
+ {
+ final StringBuilder record = new StringBuilder();
+ for (int i = 0; i < kinds.length; i++) {
+ if (i > 0) {
+ record.append(',');
+ }
+ record.append(csvEscape(converter.convert(kinds[i], readValue(rs, i + 1,
kinds[i]))));
+ }
+ writer.write(record.append('\n').toString());
+ }
+
+ /**
+ * Escapes a value as one CSV field, as per RFC 4180.
+ */
+ static String csvEscape(@Nullable final String value)
+ {
+ if (value == null) {
+ return "";
+ } else if (value.isEmpty()) {
+ return "\"\"";
+ } else if (value.contains(",") || value.contains("\"") ||
value.contains("\n") || value.contains("\r")
+ // Derby strips leading and trailing spaces from an unquoted
field, which would silently corrupt a
+ // value such as a datasource name with a boundary space
+ || Character.isWhitespace(value.charAt(0))
+ || Character.isWhitespace(value.charAt(value.length() - 1))) {
+ return "\"" + StringUtils.replace(value, "\"", "\"\"") + "\"";
+ } else {
+ return value;
+ }
+ }
+}
diff --git a/services/src/test/java/org/apache/druid/cli/CreateTablesTest.java
b/services/src/test/java/org/apache/druid/cli/CreateTablesTest.java
index 7a1e7cd668a..2c8a382bede 100644
--- a/services/src/test/java/org/apache/druid/cli/CreateTablesTest.java
+++ b/services/src/test/java/org/apache/druid/cli/CreateTablesTest.java
@@ -61,6 +61,7 @@ public class CreateTablesTest
Assertions.assertFalse(tableExists(config.getAuditTable()));
Assertions.assertFalse(tableExists(config.getSupervisorTable()));
Assertions.assertFalse(tableExists(config.getTaskLockTable()));
+ Assertions.assertFalse(tableExists(config.getIndexingStatesTable()));
// Run CreateTables
CreateTables createTables = new CreateTables()
@@ -85,6 +86,7 @@ public class CreateTablesTest
Assertions.assertTrue(tableExists(config.getAuditTable()));
Assertions.assertTrue(tableExists(config.getSupervisorTable()));
Assertions.assertTrue(tableExists(config.getTaskLockTable()));
+ Assertions.assertTrue(tableExists(config.getIndexingStatesTable()));
}
private boolean tableExists(String tableName)
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 00000000000..638e7c45230
--- /dev/null
+++ b/services/src/test/java/org/apache/druid/cli/ExportMetadataTest.java
@@ -0,0 +1,576 @@
+/*
+ * 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.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.io.BaseEncoding;
+import com.opencsv.CSVReader;
+import com.opencsv.CSVReaderBuilder;
+import com.opencsv.RFC4180ParserBuilder;
+import com.opencsv.enums.CSVReaderNullFieldIndicator;
+import com.opencsv.exceptions.CsvValidationException;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.metadata.TestDerbyConnector;
+import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
+import org.apache.druid.testing.TemporaryFolderExtension;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * Tests the export of the segments table, which the {@code export-metadata}
tool reads from the database and writes
+ * as CSV in a single pass.
+ */
+public class ExportMetadataTest
+{
+ private static final String SEGMENTS_TABLE = "druid_segments";
+ private static final String INDEXING_STATES_TABLE = "druid_indexingStates";
+ private static final String PAYLOAD_JSON =
"{\"type\":\"test\",\"path\":\"C:\\\\druid\\\\segments\"}";
+ /** The nine columns of a segments table of any Druid version. {@code end}
is a reserved word in Derby. */
+ private static final String SEGMENTS_BASE_COLUMNS =
+ "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";
+ private static final String INSERT_LEGACY_SEGMENT = "INSERT INTO
druid_segments VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ private static final List<String> SEGMENTS_COLUMN_ORDER =
ExportMetadata.SEGMENTS.columnOrder;
+
+ @RegisterExtension
+ public final TemporaryFolderExtension tempFolder =
TemporaryFolderExtension.testCaseScoped();
+
+ @RegisterExtension
+ public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule =
+ new TestDerbyConnector.DerbyConnectorRule();
+
+ private ExportMetadata tool;
+ private File outputDir;
+
+ @BeforeEach
+ public void setUp() throws IOException
+ {
+ ExportMetadata.configureJsonMapper();
+ outputDir = tempFolder.newFolder("export");
+ tool = new ExportMetadata();
+ tool.outputPath = outputDir.getAbsolutePath();
+ }
+
+ @Test
+ public void testOrderColumns()
+ {
+ // Columns as reported by a table where the newer columns were added by
ALTER TABLE in arbitrary order
+ Assertions.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.orderColumns(SEGMENTS_COLUMN_ORDER, 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"
+ ), ImmutableSet.of())
+ );
+
+ // Ordering ignores case, and columns missing from the source table are
skipped
+ Assertions.assertEquals(
+ ImmutableList.of("ID", "DATASOURCE", "CREATED_DATE", "START", "END",
"PARTITIONED", "VERSION", "USED",
+ "PAYLOAD"),
+ ExportMetadata.orderColumns(SEGMENTS_COLUMN_ORDER, ImmutableList.of(
+ "PAYLOAD", "USED", "ID", "DATASOURCE", "CREATED_DATE", "START",
"END", "PARTITIONED", "VERSION"
+ ), ImmutableSet.of())
+ );
+
+ // Unknown columns are appended at the end, in the order reported by the
database
+ Assertions.assertEquals(
+ ImmutableList.of("id", "payload", "custom_col", "another_col"),
+ ExportMetadata.orderColumns(
+ SEGMENTS_COLUMN_ORDER,
+ ImmutableList.of("custom_col", "id", "payload", "another_col"),
+ ImmutableSet.of()
+ )
+ );
+
+ // Excluded columns are left out, whether they are listed in the canonical
order or not
+ Assertions.assertEquals(
+ ImmutableList.of("payload", "custom_col"),
+ ExportMetadata.orderColumns(
+ SEGMENTS_COLUMN_ORDER,
+ ImmutableList.of("custom_col", "ID", "payload", "another_col"),
+ ImmutableSet.of("id", "another_col")
+ )
+ );
+ }
+
+ /**
+ * The segments table is exported in the canonical column order, with all of
its columns, whatever the physical
+ * column order of the source table is.
+ */
+ @Test
+ public void testExportSegments() throws IOException, CsvValidationException
+ {
+ // The optional columns are declared in an order which differs from the
canonical one
+ createSegmentsTable(
+ "upgraded_from_segment_id VARCHAR(255)",
+ "num_rows BIGINT",
+ "used_status_last_updated VARCHAR(255)",
+ "indexing_state_fingerprint VARCHAR(255)"
+ );
+ insert(
+ "INSERT INTO druid_segments (id, dataSource, created_date, start,
\"END\", partitioned, version, used, "
+ + "payload, upgraded_from_segment_id, num_rows,
used_status_last_updated, indexing_state_fingerprint) "
+ + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ "seg1",
+ "my_datasource",
+ "2024-01-15",
+ "2024-01-01",
+ "2024-01-02",
+ true,
+ "v1",
+ true,
+ StringUtils.toUtf8(PAYLOAD_JSON),
+ "upgraded_seg_0",
+ 42L,
+ "2024-06-01T00:00:00.000Z",
+ "fp_abc123"
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertEquals(13, record.length);
+ Assertions.assertEquals("seg1", record[0]);
+ Assertions.assertEquals("my_datasource", record[1]);
+ Assertions.assertEquals("2024-01-15", record[2]);
+ Assertions.assertEquals("2024-01-01", record[3]);
+ Assertions.assertEquals("2024-01-02", record[4]);
+ // Booleans are written as 1 and 0
+ Assertions.assertEquals("1", record[5]);
+ Assertions.assertEquals("v1", record[6]);
+ Assertions.assertEquals("1", record[7]);
+ // The payload is written as the JSON it holds, with its backslashes
preserved
+ Assertions.assertEquals(PAYLOAD_JSON, record[8]);
+ // The optional columns follow in the canonical order, not in the order of
the source table
+ Assertions.assertEquals("2024-06-01T00:00:00.000Z", record[9]);
+ Assertions.assertEquals("fp_abc123", record[10]);
+ Assertions.assertEquals("upgraded_seg_0", record[11]);
+ Assertions.assertEquals("42", record[12]);
+ }
+
+ /**
+ * A table with only the nine columns of an older Druid version is exported
with those nine columns.
+ */
+ @Test
+ public void testExportSegmentsWithLegacyColumns() throws IOException,
CsvValidationException
+ {
+ createSegmentsTable();
+ insert(
+ INSERT_LEGACY_SEGMENT,
+ "old_seg",
+ "old_ds",
+ "2020-01-01",
+ "2020-01-01",
+ "2020-01-02",
+ false,
+ "v0",
+ true,
+ StringUtils.toUtf8(PAYLOAD_JSON)
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertEquals(9, record.length);
+ Assertions.assertEquals("old_seg", record[0]);
+ Assertions.assertEquals("0", record[5]);
+ Assertions.assertEquals("1", record[7]);
+ Assertions.assertEquals(PAYLOAD_JSON, record[8]);
+ }
+
+ /**
+ * Values containing commas, double quotes, backslashes and line breaks are
escaped as per RFC 4180, and a record
+ * spanning several lines is written as a single record.
+ */
+ @Test
+ public void testExportSegmentsWithSpecialCharacters() throws IOException,
CsvValidationException
+ {
+ final String id = "seg,id\\1";
+ final String datasource = "ds\nwith\nnewlines";
+ final String version = "v\"quoted\"";
+ final String fingerprint = "line1\rline2,line3";
+
+ createSegmentsTable("indexing_state_fingerprint VARCHAR(255)");
+ insert(
+ "INSERT INTO druid_segments VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ id,
+ datasource,
+ "2024-01-15",
+ "2024-01-01",
+ "2024-01-02",
+ true,
+ version,
+ true,
+ StringUtils.toUtf8(PAYLOAD_JSON),
+ fingerprint
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertEquals(10, record.length);
+ Assertions.assertEquals(id, record[0]);
+ Assertions.assertEquals(datasource, record[1]);
+ Assertions.assertEquals(version, record[6]);
+ Assertions.assertEquals(PAYLOAD_JSON, record[8]);
+ Assertions.assertEquals(fingerprint, record[9]);
+ }
+
+ /**
+ * A NULL is written as an unquoted empty field and an empty string as a
quoted one, so that a nullable column such
+ * as {@code num_rows} is imported as a NULL rather than as an empty string.
+ */
+ @Test
+ public void testExportSegmentsDistinguishesNullFromEmptyString() throws
IOException, CsvValidationException
+ {
+ createSegmentsTable(
+ "used_status_last_updated VARCHAR(255)",
+ "indexing_state_fingerprint VARCHAR(255)",
+ "num_rows BIGINT"
+ );
+ insert(
+ "INSERT INTO druid_segments (id, dataSource, created_date, start,
\"END\", partitioned, version, used, "
+ + "payload, used_status_last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?,
?, ?)",
+ "seg1",
+ "my_datasource",
+ "2024-01-15",
+ "2024-01-01",
+ "2024-01-02",
+ true,
+ "v1",
+ true,
+ StringUtils.toUtf8(PAYLOAD_JSON),
+ ""
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertEquals(12, record.length);
+ Assertions.assertEquals("", record[9]);
+ Assertions.assertNull(record[10]);
+ Assertions.assertNull(record[11]);
+ }
+
+ /**
+ * With {@code --use-hex-blobs} and {@code --booleans-as-strings}, the
payload stays hex-encoded and the booleans
+ * are written as true/false.
+ */
+ @Test
+ public void testExportSegmentsWithHexBlobsAndBooleanStrings() throws
IOException, CsvValidationException
+ {
+ tool.useHexBlobs = true;
+ tool.booleansAsStrings = true;
+
+ createSegmentsTable();
+ insert(
+ INSERT_LEGACY_SEGMENT,
+ "seg1",
+ "my_datasource",
+ "2024-01-15",
+ "2024-01-01",
+ "2024-01-02",
+ true,
+ "v1",
+ false,
+ StringUtils.toUtf8(PAYLOAD_JSON)
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertEquals("true", record[5]);
+ Assertions.assertEquals("false", record[7]);
+
Assertions.assertEquals(BaseEncoding.base16().encode(StringUtils.toUtf8(PAYLOAD_JSON)),
record[8]);
+ }
+
+ /**
+ * With a deep storage migration option, the load spec of every segment
payload is rewritten.
+ */
+ @Test
+ public void testExportSegmentsRewritesLoadSpec() throws IOException,
CsvValidationException
+ {
+ tool.newLocalPath = "/new/local/path";
+
+ final String payload = "{\"dataSource\":\"my_datasource\","
+ +
"\"interval\":\"2024-01-01T00:00:00.000Z/2024-01-02T00:00:00.000Z\","
+ + "\"version\":\"v1\","
+ +
"\"loadSpec\":{\"type\":\"local\",\"path\":\"/old/path/index.zip\"},"
+ +
"\"dimensions\":\"\",\"metrics\":\"\",\"shardSpec\":{\"type\":\"numbered\","
+ +
"\"partitionNum\":0,\"partitions\":1},\"binaryVersion\":9,\"size\":100,"
+ + "\"identifier\":\"seg1\"}";
+
+ createSegmentsTable();
+ insert(
+ INSERT_LEGACY_SEGMENT,
+ "seg1",
+ "my_datasource",
+ "2024-01-15",
+ "2024-01-01",
+ "2024-01-02",
+ true,
+ "v1",
+ true,
+ StringUtils.toUtf8(payload)
+ );
+
+ final String[] record = exportSegments().get(0);
+
+ Assertions.assertTrue(
+
record[8].contains("\"loadSpec\":{\"type\":\"local\",\"path\":\"/new/local/path/"),
+ record[8]
+ );
+ }
+
+ /**
+ * The table backing {@code schema_fingerprint} is exported, without its
generated {@code id}, which the target
+ * database assigns on import.
+ */
+ @Test
+ public void testExportSegmentSchemas() throws IOException,
CsvValidationException
+ {
+ createTable(
+ "druid_segmentSchemas",
+ "id BIGINT GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1)
NOT NULL, "
+ + "created_date VARCHAR(255) NOT NULL, datasource VARCHAR(255) NOT
NULL, "
+ + "fingerprint VARCHAR(255) NOT NULL, payload BLOB NOT NULL, used
BOOLEAN NOT NULL, "
+ + "used_status_last_updated VARCHAR(255) NOT NULL, version INTEGER NOT
NULL, PRIMARY KEY (id)"
+ );
+ insert(
+ "INSERT INTO druid_segmentSchemas (created_date, datasource,
fingerprint, payload, used, "
+ + "used_status_last_updated, version) VALUES (?, ?, ?, ?, ?, ?, ?)",
+ "2024-01-15",
+ "my_datasource",
+ "fp_schema_1",
+ StringUtils.toUtf8(PAYLOAD_JSON),
+ true,
+ "2024-06-01",
+ 1
+ );
+
+ exportTable("druid_segmentSchemas", ExportMetadata.SEGMENT_SCHEMAS,
tool::convertValue);
+ final String[] record = readRecords("druid_segmentSchemas").get(0);
+
+ // The generated id is not exported, and the remaining columns are written
in the canonical order
+ Assertions.assertEquals(7, record.length);
+ Assertions.assertEquals("fp_schema_1", record[0]);
+ Assertions.assertEquals("2024-01-15", record[1]);
+ Assertions.assertEquals("my_datasource", record[2]);
+ Assertions.assertEquals(PAYLOAD_JSON, record[3]);
+ Assertions.assertEquals("1", record[4]);
+ Assertions.assertEquals("2024-06-01", record[5]);
+ Assertions.assertEquals("1", record[6]);
+ }
+
+ /**
+ * The table backing {@code indexing_state_fingerprint} is exported.
+ */
+ @Test
+ public void testExportIndexingStates() throws IOException,
CsvValidationException
+ {
+ createTable(
+ INDEXING_STATES_TABLE,
+ "created_date VARCHAR(255) NOT NULL, dataSource VARCHAR(255) NOT NULL,
"
+ + "fingerprint VARCHAR(255) NOT NULL, payload BLOB NOT NULL, used
BOOLEAN NOT NULL, "
+ + "pending BOOLEAN NOT NULL, used_status_last_updated VARCHAR(255) NOT
NULL, PRIMARY KEY (fingerprint)"
+ );
+ insert(
+ "INSERT INTO druid_indexingStates VALUES (?, ?, ?, ?, ?, ?, ?)",
+ "2024-01-15",
+ "my_datasource",
+ "fp_state_1",
+ StringUtils.toUtf8(PAYLOAD_JSON),
+ true,
+ false,
+ "2024-06-01"
+ );
+
+ exportTable(INDEXING_STATES_TABLE, ExportMetadata.INDEXING_STATES,
tool::convertValue);
+ final String[] record = readRecords(INDEXING_STATES_TABLE).get(0);
+
+ Assertions.assertEquals(7, record.length);
+ Assertions.assertEquals("fp_state_1", record[0]);
+ Assertions.assertEquals("2024-01-15", record[1]);
+ Assertions.assertEquals("my_datasource", record[2]);
+ Assertions.assertEquals(PAYLOAD_JSON, record[3]);
+ Assertions.assertEquals("1", record[4]);
+ Assertions.assertEquals("0", record[5]);
+ Assertions.assertEquals("2024-06-01", record[6]);
+ }
+
+ /**
+ * A table which the metadata store does not have, such as the segment
schemas of a store without centralized
+ * datasource schema, is skipped instead of failing the export, and the file
of an earlier export into the same
+ * directory is removed rather than left behind as stale data.
+ */
+ @Test
+ public void testExportSkipsMissingOptionalTable() throws IOException
+ {
+ final Path staleFile =
outputDir.toPath().resolve("DRUID_SEGMENTSCHEMAS.csv");
+ Files.write(staleFile, StringUtils.toUtf8("stale,data\n"));
+
+ exportTable("druid_segmentSchemas", ExportMetadata.SEGMENT_SCHEMAS,
tool::convertValue);
+
+ Assertions.assertFalse(Files.exists(staleFile));
+ }
+
+ /**
+ * A failed column lookup must not look like a table the metadata store does
not have, which would silently skip
+ * an optional table.
+ */
+ @Test
+ public void testExportFailsWhenColumnLookupFails()
+ {
+ createTable(INDEXING_STATES_TABLE, "fingerprint VARCHAR(255) NOT NULL,
payload BLOB NOT NULL");
+
+ final TestDerbyConnector connector = new TestDerbyConnector(
+ derbyConnectorRule.getMetadataConnectorConfig(),
+ derbyConnectorRule.metadataTablesConfigSupplier().get(),
+ derbyConnectorRule.getConnector().getJdbcUri(),
+ CentralizedDatasourceSchemaConfig.create()
+ )
+ {
+ @Override
+ public String getMetadataTableSchema(final Connection connection) throws
SQLException
+ {
+ throw new SQLException("Could not read the database metadata");
+ }
+ };
+
+ final Exception e = Assertions.assertThrows(
+ Exception.class,
+ () -> tool.exportTable(
+ new MetadataCsvExporter(connector),
+ connector,
+ StringUtils.toUpperCase(INDEXING_STATES_TABLE),
+ ExportMetadata.INDEXING_STATES,
+ tool::convertValue
+ )
+ );
+ Assertions.assertInstanceOf(SQLException.class,
Throwables.getRootCause(e));
+ }
+
+ @Test
+ public void testExportFailsWhenTableIsMissing()
+ {
+ final ISE e = Assertions.assertThrows(
+ ISE.class,
+ this::exportSegmentsTable
+ );
+ Assertions.assertTrue(e.getMessage().contains("does not exist in this
metadata store"), e.getMessage());
+ }
+
+ /**
+ * Creates the segments table with the nine columns every version has,
followed by the given optional columns.
+ */
+ private void createSegmentsTable(final String... optionalColumns)
+ {
+ final StringBuilder columns = new StringBuilder(SEGMENTS_BASE_COLUMNS);
+ for (String column : optionalColumns) {
+ columns.append(", ").append(column);
+ }
+ createTable(SEGMENTS_TABLE, columns.append(", PRIMARY
KEY(id)").toString());
+ }
+
+ private void createTable(final String tableName, final String columns)
+ {
+ derbyConnectorRule.getConnector().getDBI().withHandle(handle -> {
+ handle.execute(StringUtils.format("CREATE TABLE %s (%s)", tableName,
columns));
+ return null;
+ });
+ }
+
+ private void insert(final String sql, final Object... values)
+ {
+ derbyConnectorRule.getConnector().getDBI().withHandle(handle -> {
+ handle.execute(sql, values);
+ return null;
+ });
+ }
+
+ private void exportSegmentsTable()
+ {
+ exportTable(SEGMENTS_TABLE, ExportMetadata.SEGMENTS,
tool::convertSegmentValue);
+ }
+
+ private void exportTable(
+ final String tableName,
+ final ExportMetadata.TableSpec spec,
+ final MetadataCsvExporter.ValueConverter converter
+ )
+ {
+ final TestDerbyConnector connector = derbyConnectorRule.getConnector();
+ tool.exportTable(
+ new MetadataCsvExporter(connector),
+ connector,
+ StringUtils.toUpperCase(tableName),
+ spec,
+ converter
+ );
+ }
+
+ /**
+ * Exports the segments table and reads the written CSV back as records, so
that a record spanning several lines is
+ * read as one record and a NULL stays distinguishable from an empty string.
+ */
+ private List<String[]> exportSegments() throws IOException,
CsvValidationException
+ {
+ exportSegmentsTable();
+ return readRecords(SEGMENTS_TABLE);
+ }
+
+ private List<String[]> readRecords(final String tableName) throws
IOException, CsvValidationException
+ {
+ final Path outputFile =
outputDir.toPath().resolve(StringUtils.toUpperCase(tableName) + ".csv");
+ final List<String[]> records = new ArrayList<>();
+ try (
+ CSVReader reader = new
CSVReaderBuilder(Files.newBufferedReader(outputFile, StandardCharsets.UTF_8))
+ .withCSVParser(
+ new
RFC4180ParserBuilder().withFieldAsNull(CSVReaderNullFieldIndicator.EMPTY_SEPARATORS).build()
+ )
+ .withKeepCarriageReturn(true)
+ .build()
+ ) {
+ String[] record;
+ while ((record = reader.readNext()) != null) {
+ records.add(record);
+ }
+ }
+ records.sort(Comparator.comparing(record -> record[0]));
+ return records;
+ }
+}
diff --git
a/services/src/test/java/org/apache/druid/cli/MetadataCsvExporterTest.java
b/services/src/test/java/org/apache/druid/cli/MetadataCsvExporterTest.java
new file mode 100644
index 00000000000..1105d53c886
--- /dev/null
+++ b/services/src/test/java/org/apache/druid/cli/MetadataCsvExporterTest.java
@@ -0,0 +1,250 @@
+/*
+ * 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 org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.metadata.TestDerbyConnector;
+import org.apache.druid.testing.TemporaryFolderExtension;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+/**
+ * Tests the CSV export of {@link MetadataCsvExporter} against Derby,
exercising the generic JDBC path which is used
+ * for every metadata store.
+ */
+public class MetadataCsvExporterTest
+{
+ private static final MetadataCsvExporter.ValueConverter KEEP_VALUE = (kind,
value) -> value;
+
+ @RegisterExtension
+ public final TemporaryFolderExtension tempFolder =
TemporaryFolderExtension.testCaseScoped();
+
+ @RegisterExtension
+ public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule =
+ new TestDerbyConnector.DerbyConnectorRule();
+
+ @Test
+ public void testCsvEscape()
+ {
+ Assertions.assertEquals("hello", MetadataCsvExporter.csvEscape("hello"));
+ Assertions.assertEquals("\"value,with,commas\"",
MetadataCsvExporter.csvEscape("value,with,commas"));
+ Assertions.assertEquals("\"value\"\"with\"\"quotes\"",
MetadataCsvExporter.csvEscape("value\"with\"quotes"));
+ Assertions.assertEquals("\"line1\nline2\"",
MetadataCsvExporter.csvEscape("line1\nline2"));
+ Assertions.assertEquals("\"line1\rline2\"",
MetadataCsvExporter.csvEscape("line1\rline2"));
+ // A backslash is an ordinary character
+ Assertions.assertEquals("back\\slash",
MetadataCsvExporter.csvEscape("back\\slash"));
+ // Derby strips the boundary spaces of an unquoted field, so such a value
is quoted
+ Assertions.assertEquals("\" leading\"", MetadataCsvExporter.csvEscape("
leading"));
+ Assertions.assertEquals("\"trailing \"",
MetadataCsvExporter.csvEscape("trailing "));
+ Assertions.assertEquals("no boundary space",
MetadataCsvExporter.csvEscape("no boundary space"));
+ // A NULL is an empty field, while an empty string is a quoted empty
field, so that the two stay distinct
+ Assertions.assertEquals("", MetadataCsvExporter.csvEscape(null));
+ Assertions.assertEquals("\"\"", MetadataCsvExporter.csvEscape(""));
+ }
+
+ @Test
+ public void testExportTable() throws IOException
+ {
+ createTable(
+ "test_export",
+ "name VARCHAR(255) NOT NULL, payload BLOB NOT NULL, active BOOLEAN NOT
NULL, PRIMARY KEY(name)"
+ );
+ insert("test_export", "key1", StringUtils.toUtf8("{\"type\":\"test\"}"),
true);
+ insert("test_export", "key2", StringUtils.toUtf8("{\"value\":42}"), false);
+
+ final List<String> lines = export("TEST_EXPORT", ImmutableList.of("NAME",
"PAYLOAD", "ACTIVE"));
+
+ // Binary values are hex-encoded and booleans are written as true/false
+ Assertions.assertEquals(
+ ImmutableList.of(
+ "key1," +
BaseEncoding.base16().encode(StringUtils.toUtf8("{\"type\":\"test\"}")) +
",true",
+ "key2," +
BaseEncoding.base16().encode(StringUtils.toUtf8("{\"value\":42}")) + ",false"
+ ),
+ lines
+ );
+ }
+
+ @Test
+ public void testExportTableEscapesSpecialCharacters() throws IOException
+ {
+ createTable(
+ "test_special",
+ "name VARCHAR(255) NOT NULL, description VARCHAR(1024), PRIMARY
KEY(name)"
+ );
+ insert("test_special", "commas", "value,with,commas");
+ insert("test_special", "quotes", "value\"with\"quotes");
+ insert("test_special", "backslash", "back\\slash");
+ insert("test_special", "simple", "plain_value");
+
+ final List<String> lines = export("TEST_SPECIAL", ImmutableList.of("NAME",
"DESCRIPTION"));
+
+ Assertions.assertEquals(
+ ImmutableList.of(
+ "backslash,back\\slash",
+ "commas,\"value,with,commas\"",
+ "quotes,\"value\"\"with\"\"quotes\"",
+ "simple,plain_value"
+ ),
+ lines
+ );
+ }
+
+ @Test
+ public void testExportTableDistinguishesNullFromEmptyString() throws
IOException
+ {
+ createTable(
+ "test_nulls",
+ "name VARCHAR(255) NOT NULL, payload BLOB, description VARCHAR(255),
PRIMARY KEY(name)"
+ );
+ execute("INSERT INTO test_nulls (name) VALUES ('all_null')");
+ execute("INSERT INTO test_nulls (name, description) VALUES
('empty_string', '')");
+
+ final List<String> lines = export("TEST_NULLS", ImmutableList.of("NAME",
"PAYLOAD", "DESCRIPTION"));
+
+ // A NULL is an unquoted empty field, an empty string a quoted one
+ Assertions.assertEquals(
+ ImmutableList.of("all_null,,", "empty_string,,\"\""),
+ lines
+ );
+ }
+
+ @Test
+ public void testExportTableWithMultilineValues() throws IOException
+ {
+ createTable(
+ "test_multiline",
+ "name VARCHAR(255) NOT NULL, description VARCHAR(1024), PRIMARY
KEY(name)"
+ );
+ insert("test_multiline", "breaks", "line1\nline2\rline3");
+
+ final Path outputFile = exportToFile("TEST_MULTILINE",
ImmutableList.of("NAME", "DESCRIPTION"), KEEP_VALUE);
+
+ // A value containing a line break is quoted, so that the record can be
read back as one record
+ Assertions.assertEquals(
+ "breaks,\"line1\nline2\rline3\"\n",
+ new String(Files.readAllBytes(outputFile), StandardCharsets.UTF_8)
+ );
+ }
+
+ @Test
+ public void testExportTableInGivenColumnOrder() throws IOException
+ {
+ // "end" is a reserved word, so it must be quoted in the export query
+ createTable(
+ "test_column_order",
+ "id VARCHAR(255) NOT NULL, used_status_last_updated VARCHAR(255),
\"END\" VARCHAR(255), "
+ + "used BOOLEAN NOT NULL, PRIMARY KEY(id)"
+ );
+ insert("test_column_order", "seg1", "2024-01-01", "2024-01-02", true);
+
+ final List<String> lines =
+ export("TEST_COLUMN_ORDER", ImmutableList.of("ID", "END", "USED",
"USED_STATUS_LAST_UPDATED"));
+
+
Assertions.assertEquals(ImmutableList.of("seg1,2024-01-02,true,2024-01-01"),
lines);
+ }
+
+ @Test
+ public void testExportTableAppliesValueConverter() throws IOException
+ {
+ createTable(
+ "test_converter",
+ "name VARCHAR(255) NOT NULL, payload BLOB NOT NULL, active BOOLEAN NOT
NULL, PRIMARY KEY(name)"
+ );
+ insert("test_converter", "key1",
StringUtils.toUtf8("{\"type\":\"test\"}"), true);
+
+ final Path outputFile = exportToFile(
+ "TEST_CONVERTER",
+ ImmutableList.of("NAME", "PAYLOAD", "ACTIVE"),
+ (kind, value) -> {
+ switch (kind) {
+ case BINARY:
+ return StringUtils.fromUtf8(BaseEncoding.base16().decode(value));
+ case BOOLEAN:
+ return "true".equals(value) ? "1" : "0";
+ default:
+ return value;
+ }
+ }
+ );
+
+ Assertions.assertEquals(
+ ImmutableList.of("key1,\"{\"\"type\"\":\"\"test\"\"}\",1"),
+ Files.readAllLines(outputFile, StandardCharsets.UTF_8)
+ );
+ }
+
+ private void createTable(final String tableName, final String columns)
+ {
+ execute(StringUtils.format("CREATE TABLE %s (%s)", tableName, columns));
+ }
+
+ private void execute(final String sql)
+ {
+ derbyConnectorRule.getConnector().getDBI().withHandle(handle -> {
+ handle.execute(sql);
+ return null;
+ });
+ }
+
+ private void insert(final String tableName, final Object... values)
+ {
+ final StringBuilder placeholders = new StringBuilder();
+ for (int i = 0; i < values.length; i++) {
+ placeholders.append(i == 0 ? "?" : ",?");
+ }
+ derbyConnectorRule.getConnector().getDBI().withHandle(handle -> {
+ handle.execute(StringUtils.format("INSERT INTO %s VALUES (%s)",
tableName, placeholders), values);
+ return null;
+ });
+ }
+
+ /**
+ * Exports the given table and returns its records, sorted so that the
assertions do not depend on the order in
+ * which the database returns the rows.
+ */
+ private List<String> export(final String tableName, final List<String>
columns) throws IOException
+ {
+ final List<String> lines =
+ Files.readAllLines(exportToFile(tableName, columns, KEEP_VALUE),
StandardCharsets.UTF_8);
+ lines.sort(String::compareTo);
+ return lines;
+ }
+
+ private Path exportToFile(
+ final String tableName,
+ final List<String> columns,
+ final MetadataCsvExporter.ValueConverter converter
+ ) throws IOException
+ {
+ final Path outputFile =
tempFolder.newFolder(tableName).toPath().resolve(tableName + ".csv");
+ new MetadataCsvExporter(derbyConnectorRule.getConnector())
+ .exportTable(tableName, columns, outputFile, converter);
+ return outputFile;
+ }
+}
diff --git a/website/.spelling b/website/.spelling
index 0689e3c47a6..a27569dcd2d 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -149,6 +149,7 @@ ISO8601
IndexSpec
IndexTask
InfluxDB
+InnoDB
InputFormat
InputSource
InputSources
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
