This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 9d33df19d9 ARROW-17631: [Java] Propagate table/columns comments into
Arrow Schema (#14081)
9d33df19d9 is described below
commit 9d33df19d9a98df5caf134f2792e5c81bca90ae3
Author: Igor Suhorukov <[email protected]>
AuthorDate: Wed Sep 14 14:59:43 2022 +0300
ARROW-17631: [Java] Propagate table/columns comments into Arrow Schema
(#14081)
Allow user to provide comment in Arrow Schema from JdbcToArrowConfig . It
will be very useful metadata in real life (medium to large scale project) for
documentation and maintenance topics. Apache Spark code use "comment" key for
such metadata, so this looks like reasonable default name for metadata in Arrow
schema too
Authored-by: igor.suhorukov <[email protected]>
Signed-off-by: David Li <[email protected]>
---
.../arrow/adapter/jdbc/JdbcToArrowConfig.java | 23 +++
.../adapter/jdbc/JdbcToArrowConfigBuilder.java | 23 +++
.../arrow/adapter/jdbc/JdbcToArrowUtils.java | 15 +-
.../jdbc/JdbcToArrowCommentMetadataTest.java | 176 +++++++++++++++++++++
.../adapter/jdbc/src/test/resources/h2/comment.sql | 21 +++
.../resources/h2/expectedSchemaWithComments.json | 51 ++++++
.../h2/expectedSchemaWithCommentsAndJdbcMeta.json | 100 ++++++++++++
7 files changed, 405 insertions(+), 4 deletions(-)
diff --git
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfig.java
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfig.java
index b475ee046b..d11682c7d4 100644
---
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfig.java
+++
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfig.java
@@ -60,6 +60,8 @@ public final class JdbcToArrowConfig {
private final Map<String, JdbcFieldInfo> arraySubTypesByColumnName;
private final Map<Integer, JdbcFieldInfo> explicitTypesByColumnIndex;
private final Map<String, JdbcFieldInfo> explicitTypesByColumnName;
+ private final Map<String, String> schemaMetadata;
+ private final Map<Integer, Map<String, String>> columnMetadataByColumnIndex;
private final RoundingMode bigDecimalRoundingMode;
/**
* The maximum rowCount to read each time when partially convert data.
@@ -174,6 +176,8 @@ public final class JdbcToArrowConfig {
jdbcToArrowTypeConverter,
null,
null,
+ null,
+ null,
bigDecimalRoundingMode);
}
@@ -188,6 +192,8 @@ public final class JdbcToArrowConfig {
Function<JdbcFieldInfo, ArrowType> jdbcToArrowTypeConverter,
Map<Integer, JdbcFieldInfo> explicitTypesByColumnIndex,
Map<String, JdbcFieldInfo> explicitTypesByColumnName,
+ Map<String, String> schemaMetadata,
+ Map<Integer, Map<String, String>> columnMetadataByColumnIndex,
RoundingMode bigDecimalRoundingMode) {
Preconditions.checkNotNull(allocator, "Memory allocator cannot be null");
this.allocator = allocator;
@@ -199,6 +205,8 @@ public final class JdbcToArrowConfig {
this.targetBatchSize = targetBatchSize;
this.explicitTypesByColumnIndex = explicitTypesByColumnIndex;
this.explicitTypesByColumnName = explicitTypesByColumnName;
+ this.schemaMetadata = schemaMetadata;
+ this.columnMetadataByColumnIndex = columnMetadataByColumnIndex;
this.bigDecimalRoundingMode = bigDecimalRoundingMode;
// set up type converter
@@ -312,6 +320,21 @@ public final class JdbcToArrowConfig {
}
}
+ /**
+ * Return schema level metadata or null if not provided.
+ */
+ public Map<String, String> getSchemaMetadata() {
+ return schemaMetadata;
+ }
+
+ /**
+ * Return metadata from columnIndex->meta map on per field basis
+ * or null if not provided.
+ */
+ public Map<Integer, Map<String, String>> getColumnMetadataByColumnIndex() {
+ return columnMetadataByColumnIndex;
+ }
+
public RoundingMode getBigDecimalRoundingMode() {
return bigDecimalRoundingMode;
}
diff --git
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfigBuilder.java
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfigBuilder.java
index 5618087669..2fe0492deb 100644
---
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfigBuilder.java
+++
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowConfigBuilder.java
@@ -40,6 +40,8 @@ public class JdbcToArrowConfigBuilder {
private Map<String, JdbcFieldInfo> arraySubTypesByColumnName;
private Map<Integer, JdbcFieldInfo> explicitTypesByColumnIndex;
private Map<String, JdbcFieldInfo> explicitTypesByColumnName;
+ private Map<String, String> schemaMetadata;
+ private Map<Integer, Map<String, String>> columnMetadataByColumnIndex;
private int targetBatchSize;
private Function<JdbcFieldInfo, ArrowType> jdbcToArrowTypeConverter;
private RoundingMode bigDecimalRoundingMode;
@@ -58,6 +60,8 @@ public class JdbcToArrowConfigBuilder {
this.arraySubTypesByColumnName = null;
this.explicitTypesByColumnIndex = null;
this.explicitTypesByColumnName = null;
+ this.schemaMetadata = null;
+ this.columnMetadataByColumnIndex = null;
this.bigDecimalRoundingMode = null;
}
@@ -226,6 +230,23 @@ public class JdbcToArrowConfigBuilder {
return this;
}
+ /**
+ * Set metadata for schema.
+ */
+ public JdbcToArrowConfigBuilder setSchemaMetadata(Map<String, String>
schemaMetadata) {
+ this.schemaMetadata = schemaMetadata;
+ return this;
+ }
+
+ /**
+ * Set metadata from columnIndex->meta map on per field basis.
+ */
+ public JdbcToArrowConfigBuilder setColumnMetadataByColumnIndex(
+ Map<Integer, Map<String, String>> columnMetadataByColumnIndex) {
+ this.columnMetadataByColumnIndex = columnMetadataByColumnIndex;
+ return this;
+ }
+
/**
* Set the rounding mode used when the scale of the actual value does not
match the declared scale.
* <p>
@@ -255,6 +276,8 @@ public class JdbcToArrowConfigBuilder {
jdbcToArrowTypeConverter,
explicitTypesByColumnIndex,
explicitTypesByColumnName,
+ schemaMetadata,
+ columnMetadataByColumnIndex,
bigDecimalRoundingMode);
}
}
diff --git
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowUtils.java
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowUtils.java
index 43fed849a2..e782efde3d 100644
---
a/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowUtils.java
+++
b/java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowUtils.java
@@ -244,6 +244,8 @@ public class JdbcToArrowUtils {
for (int i = 1; i <= columnCount; i++) {
final String columnName = rsmd.getColumnLabel(i);
+ final Map<String, String> columnMetadata =
config.getColumnMetadataByColumnIndex() != null ?
+ config.getColumnMetadataByColumnIndex().get(i) : null;
final Map<String, String> metadata;
if (config.shouldIncludeMetadata()) {
metadata = new HashMap<>();
@@ -251,9 +253,15 @@ public class JdbcToArrowUtils {
metadata.put(Constants.SQL_TABLE_NAME_KEY, rsmd.getTableName(i));
metadata.put(Constants.SQL_COLUMN_NAME_KEY, columnName);
metadata.put(Constants.SQL_TYPE_KEY, rsmd.getColumnTypeName(i));
-
+ if (columnMetadata != null && !columnMetadata.isEmpty()) {
+ metadata.putAll(columnMetadata);
+ }
} else {
- metadata = null;
+ if (columnMetadata != null && !columnMetadata.isEmpty()) {
+ metadata = columnMetadata;
+ } else {
+ metadata = null;
+ }
}
final JdbcFieldInfo columnFieldInfo = getJdbcFieldInfoForColumn(rsmd, i,
config);
@@ -276,8 +284,7 @@ public class JdbcToArrowUtils {
fields.add(new Field(columnName, fieldType, children));
}
}
-
- return new Schema(fields, null);
+ return new Schema(fields, config.getSchemaMetadata());
}
static JdbcFieldInfo getJdbcFieldInfoForColumn(
diff --git
a/java/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/JdbcToArrowCommentMetadataTest.java
b/java/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/JdbcToArrowCommentMetadataTest.java
new file mode 100644
index 0000000000..8d3e599554
--- /dev/null
+++
b/java/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/JdbcToArrowCommentMetadataTest.java
@@ -0,0 +1,176 @@
+/*
+ * 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.arrow.adapter.jdbc;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.ObjectMapperFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.fasterxml.jackson.databind.ObjectWriter;
+
+public class JdbcToArrowCommentMetadataTest {
+
+ private static final String COMMENT = "comment"; //use this metadata key for
interoperability with Spark StructType
+ private final ObjectWriter schemaSerializer =
ObjectMapperFactory.newObjectMapper().writerWithDefaultPrettyPrinter();
+ private Connection conn = null;
+
+ /**
+ * This method creates Connection object and DB table and also populate data
into table for test.
+ *
+ * @throws SQLException on error
+ * @throws ClassNotFoundException on error
+ */
+ @Before
+ public void setUp() throws SQLException, ClassNotFoundException {
+ String url =
"jdbc:h2:mem:JdbcToArrowTest?characterEncoding=UTF-8;INIT=runscript from
'classpath:/h2/comment.sql'";
+ String driver = "org.h2.Driver";
+ Class.forName(driver);
+ conn = DriverManager.getConnection(url);
+ }
+
+ @After
+ public void tearDown() throws SQLException {
+ if (conn != null) {
+ conn.close();
+ conn = null;
+ }
+ }
+
+ @Test
+ public void schemaComment() throws Exception {
+ boolean includeMetadata = false;
+ String schemaJson =
schemaSerializer.writeValueAsString(getSchemaWithCommentFromQuery(includeMetadata));
+ String expectedSchema =
getExpectedSchema("/h2/expectedSchemaWithComments.json");
+ assertThat(schemaJson).isEqualTo(expectedSchema);
+ }
+
+ @Test
+ public void schemaCommentWithDatabaseMetadata() throws Exception {
+ boolean includeMetadata = true;
+ String schemaJson =
schemaSerializer.writeValueAsString(getSchemaWithCommentFromQuery(includeMetadata));
+ String expectedSchema =
getExpectedSchema("/h2/expectedSchemaWithCommentsAndJdbcMeta.json");
+ /* corresponding Apache Spark DDL after conversion:
+ ID BIGINT NOT NULL COMMENT 'Record identifier',
+ NAME STRING COMMENT 'Name of record',
+ COLUMN1 BOOLEAN,
+ COLUMNN INT COMMENT 'Informative description of columnN'
+ */
+ assertThat(schemaJson).isEqualTo(expectedSchema);
+ }
+
+ private Schema getSchemaWithCommentFromQuery(boolean includeMetadata) throws
SQLException {
+ DatabaseMetaData metaData = conn.getMetaData();
+ try (Statement statement = conn.createStatement()) {
+ try (ResultSet resultSet = statement.executeQuery("select * from
table1")) {
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ Map<Integer, Map<String, String>> columnCommentByColumnIndex =
getColumnComments(metaData, resultSetMetaData);
+
+ String tableName =
getTableNameFromResultSetMetaData(resultSetMetaData);
+ String tableComment = getTableComment(metaData, tableName);
+ JdbcToArrowConfig config = new JdbcToArrowConfigBuilder()
+ .setAllocator(new
RootAllocator()).setSchemaMetadata(Collections.singletonMap(COMMENT,
tableComment))
+
.setColumnMetadataByColumnIndex(columnCommentByColumnIndex).setIncludeMetadata(includeMetadata).build();
+ return JdbcToArrowUtils.jdbcToArrowSchema(resultSetMetaData, config);
+ }
+ }
+ }
+
+ private String getTableNameFromResultSetMetaData(ResultSetMetaData
resultSetMetaData) throws SQLException {
+ Set<String> tablesFromQuery = new HashSet<>();
+ for (int idx = 1, columnCount = resultSetMetaData.getColumnCount(); idx <=
columnCount; idx++) {
+ String tableName = resultSetMetaData.getTableName(idx);
+ if (tableName != null && !tableName.isEmpty()) {
+ tablesFromQuery.add(tableName);
+ }
+ }
+ if (tablesFromQuery.size() == 1) {
+ return tablesFromQuery.iterator().next();
+ }
+ throw new RuntimeException("Table metadata is absent or ambiguous");
+ }
+
+ private Map<Integer, Map<String, String>> getColumnComments(DatabaseMetaData
metaData,
+ ResultSetMetaData
resultSetMetaData) throws SQLException {
+ Map<Integer, Map<String, String>> columnCommentByColumnIndex = new
HashMap<>();
+ for (int columnIdx = 1, columnCount = resultSetMetaData.getColumnCount();
columnIdx <= columnCount; columnIdx++) {
+ String columnComment = getColumnComment(metaData,
resultSetMetaData.getTableName(columnIdx),
+ resultSetMetaData.getColumnName(columnIdx));
+ if (columnComment != null && !columnComment.isEmpty()) {
+ columnCommentByColumnIndex.put(columnIdx,
Collections.singletonMap(COMMENT, columnComment));
+ }
+ }
+ return columnCommentByColumnIndex;
+ }
+
+ private String getTableComment(DatabaseMetaData metaData, String tableName)
throws SQLException {
+ if (tableName == null || tableName.isEmpty()) {
+ return null;
+ }
+ String comment = null;
+ int rowCount = 0;
+ try (ResultSet tableMetadata = metaData.getTables("%", "%", tableName,
null)) {
+ if (tableMetadata.next()) {
+ comment = tableMetadata.getString("REMARKS");
+ rowCount++;
+ }
+ }
+ if (rowCount == 1) {
+ return comment;
+ }
+ if (rowCount > 1) {
+ throw new RuntimeException("Multiple tables found for table name");
+ }
+ throw new RuntimeException("Table comment not found");
+ }
+
+ private String getColumnComment(DatabaseMetaData metaData, String tableName,
String columnName) throws SQLException {
+ try (ResultSet tableMetadata = metaData.getColumns("%", "%", tableName,
columnName)) {
+ if (tableMetadata.next()) {
+ return tableMetadata.getString("REMARKS");
+ }
+ }
+ return null;
+ }
+
+ private String getExpectedSchema(String expectedResource) throws
java.io.IOException, java.net.URISyntaxException {
+ return new String(Files.readAllBytes(Paths.get(Objects.requireNonNull(
+
JdbcToArrowCommentMetadataTest.class.getResource(expectedResource)).toURI())),
StandardCharsets.UTF_8);
+ }
+}
diff --git a/java/adapter/jdbc/src/test/resources/h2/comment.sql
b/java/adapter/jdbc/src/test/resources/h2/comment.sql
new file mode 100644
index 0000000000..db8964fe1d
--- /dev/null
+++ b/java/adapter/jdbc/src/test/resources/h2/comment.sql
@@ -0,0 +1,21 @@
+--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.
+create table table1(
+ id bigint primary key,
+ name varchar(255),
+ column1 boolean,
+ columnN int
+ );
+
+COMMENT ON TABLE table1 IS 'This is super special table with valuable data';
+COMMENT ON COLUMN table1.id IS 'Record identifier';
+COMMENT ON COLUMN table1.name IS 'Name of record';
+COMMENT ON COLUMN table1.columnN IS 'Informative description of columnN';
\ No newline at end of file
diff --git
a/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithComments.json
b/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithComments.json
new file mode 100644
index 0000000000..cfdd00fdff
--- /dev/null
+++ b/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithComments.json
@@ -0,0 +1,51 @@
+{
+ "fields" : [ {
+ "name" : "ID",
+ "nullable" : false,
+ "type" : {
+ "name" : "int",
+ "bitWidth" : 64,
+ "isSigned" : true
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Record identifier",
+ "key" : "comment"
+ } ]
+ }, {
+ "name" : "NAME",
+ "nullable" : true,
+ "type" : {
+ "name" : "utf8"
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Name of record",
+ "key" : "comment"
+ } ]
+ }, {
+ "name" : "COLUMN1",
+ "nullable" : true,
+ "type" : {
+ "name" : "bool"
+ },
+ "children" : [ ]
+ }, {
+ "name" : "COLUMNN",
+ "nullable" : true,
+ "type" : {
+ "name" : "int",
+ "bitWidth" : 32,
+ "isSigned" : true
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Informative description of columnN",
+ "key" : "comment"
+ } ]
+ } ],
+ "metadata" : [ {
+ "value" : "This is super special table with valuable data",
+ "key" : "comment"
+ } ]
+}
\ No newline at end of file
diff --git
a/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithCommentsAndJdbcMeta.json
b/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithCommentsAndJdbcMeta.json
new file mode 100644
index 0000000000..967ce0a08e
--- /dev/null
+++
b/java/adapter/jdbc/src/test/resources/h2/expectedSchemaWithCommentsAndJdbcMeta.json
@@ -0,0 +1,100 @@
+{
+ "fields" : [ {
+ "name" : "ID",
+ "nullable" : false,
+ "type" : {
+ "name" : "int",
+ "bitWidth" : 64,
+ "isSigned" : true
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Record identifier",
+ "key" : "comment"
+ }, {
+ "value" : "TABLE1",
+ "key" : "SQL_TABLE_NAME"
+ }, {
+ "value" : "JDBCTOARROWTEST?CHARACTERENCODING=UTF-8",
+ "key" : "SQL_CATALOG_NAME"
+ }, {
+ "value" : "ID",
+ "key" : "SQL_COLUMN_NAME"
+ }, {
+ "value" : "BIGINT",
+ "key" : "SQL_TYPE"
+ } ]
+ }, {
+ "name" : "NAME",
+ "nullable" : true,
+ "type" : {
+ "name" : "utf8"
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Name of record",
+ "key" : "comment"
+ }, {
+ "value" : "TABLE1",
+ "key" : "SQL_TABLE_NAME"
+ }, {
+ "value" : "JDBCTOARROWTEST?CHARACTERENCODING=UTF-8",
+ "key" : "SQL_CATALOG_NAME"
+ }, {
+ "value" : "NAME",
+ "key" : "SQL_COLUMN_NAME"
+ }, {
+ "value" : "VARCHAR",
+ "key" : "SQL_TYPE"
+ } ]
+ }, {
+ "name" : "COLUMN1",
+ "nullable" : true,
+ "type" : {
+ "name" : "bool"
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "TABLE1",
+ "key" : "SQL_TABLE_NAME"
+ }, {
+ "value" : "JDBCTOARROWTEST?CHARACTERENCODING=UTF-8",
+ "key" : "SQL_CATALOG_NAME"
+ }, {
+ "value" : "COLUMN1",
+ "key" : "SQL_COLUMN_NAME"
+ }, {
+ "value" : "BOOLEAN",
+ "key" : "SQL_TYPE"
+ } ]
+ }, {
+ "name" : "COLUMNN",
+ "nullable" : true,
+ "type" : {
+ "name" : "int",
+ "bitWidth" : 32,
+ "isSigned" : true
+ },
+ "children" : [ ],
+ "metadata" : [ {
+ "value" : "Informative description of columnN",
+ "key" : "comment"
+ }, {
+ "value" : "TABLE1",
+ "key" : "SQL_TABLE_NAME"
+ }, {
+ "value" : "JDBCTOARROWTEST?CHARACTERENCODING=UTF-8",
+ "key" : "SQL_CATALOG_NAME"
+ }, {
+ "value" : "COLUMNN",
+ "key" : "SQL_COLUMN_NAME"
+ }, {
+ "value" : "INTEGER",
+ "key" : "SQL_TYPE"
+ } ]
+ } ],
+ "metadata" : [ {
+ "value" : "This is super special table with valuable data",
+ "key" : "comment"
+ } ]
+}
\ No newline at end of file
