Skip to content

Commit 81b9188

Browse files
chamikaramjchamikaramj
authored
Adds Delta Lake read support to Beam SQL (#39950)
1 parent 7898e7d commit 81b9188

16 files changed

Lines changed: 1304 additions & 65 deletions

File tree

‎.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
{
22
"comment": "Modify this file in a trivial way to cause this test suite to run.",
3-
"modification": 3,
3+
"modification": 1,
44
"https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
55
}

‎build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -396,6 +396,7 @@ tasks.register("sqlPreCommit") {
396396
dependsOn(":sdks:java:extensions:sql:expansion-service:build")
397397
dependsOn(":sdks:java:extensions:sql:hcatalog:build")
398398
dependsOn(":sdks:java:extensions:sql:iceberg:build")
399+
dependsOn(":sdks:java:extensions:sql:delta:build")
399400
dependsOn(":sdks:java:extensions:sql:jdbc:build")
400401
dependsOn(":sdks:java:extensions:sql:jdbc:preCommit")
401402
dependsOn(":sdks:java:extensions:sql:perf-tests:build")

‎sdks/java/extensions/sql/delta/build.gradle‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
plugins { id 'org.apache.beam.module' }
20+
21+
applyJavaNature(
22+
automaticModuleName: 'org.apache.beam.sdk.extensions.sql.meta.provider.delta',
23+
requireJavaVersion: JavaVersion.VERSION_17,
24+
)
25+
26+
description = "Apache Beam :: SDKs :: Java :: Extensions :: SQL :: Delta Lake"
27+
ext.summary = "Delta Lake table provider for Beam SQL."
28+
29+
dependencies {
30+
implementation project(":sdks:java:extensions:sql")
31+
implementation project(":sdks:java:core")
32+
implementation project(":sdks:java:managed")
33+
implementation project(":sdks:java:io:delta")
34+
implementation library.java.jackson_databind
35+
implementation library.java.jackson_core
36+
implementation library.java.slf4j_api
37+
implementation library.java.vendored_guava_32_1_2_jre
38+
implementation library.java.vendored_calcite_1_40_0
39+
// TODO(https://github.com/apache/beam/issues/21156): Determine how to build without this dependency
40+
provided "org.immutables:value:2.8.8"
41+
permitUnusedDeclared "org.immutables:value:2.8.8"
42+
permitUnusedDeclared project(":sdks:java:io:delta")
43+
permitUnusedDeclared library.java.slf4j_api
44+
45+
testImplementation library.java.joda_time
46+
testImplementation library.java.junit
47+
testImplementation library.java.hadoop_common
48+
testImplementation project(path: ":sdks:java:io:delta", configuration: "testRuntimeMigration")
49+
testImplementation project(path: ":runners:direct-java", configuration: "shadow")
50+
}

‎sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTable.java‎

Lines changed: 199 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,199 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.sdk.extensions.sql.meta.provider.delta;
19+
20+
import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
21+
22+
import com.fasterxml.jackson.core.type.TypeReference;
23+
import com.fasterxml.jackson.databind.JsonNode;
24+
import com.fasterxml.jackson.databind.node.ObjectNode;
25+
import java.util.HashMap;
26+
import java.util.List;
27+
import java.util.Map;
28+
import org.apache.beam.sdk.extensions.sql.TableUtils;
29+
import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTableFilter;
30+
import org.apache.beam.sdk.extensions.sql.meta.DefaultTableFilter;
31+
import org.apache.beam.sdk.extensions.sql.meta.SchemaBaseBeamTable;
32+
import org.apache.beam.sdk.extensions.sql.meta.Table;
33+
import org.apache.beam.sdk.managed.Managed;
34+
import org.apache.beam.sdk.values.PBegin;
35+
import org.apache.beam.sdk.values.PCollection;
36+
import org.apache.beam.sdk.values.POutput;
37+
import org.apache.beam.sdk.values.Row;
38+
import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexNode;
39+
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
40+
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
41+
import org.checkerframework.checker.nullness.qual.Nullable;
42+
43+
class DeltaTable extends SchemaBaseBeamTable {
44+
@VisibleForTesting static final String VERSION_FIELD = "version";
45+
@VisibleForTesting static final String TIMESTAMP_FIELD = "timestamp";
46+
@VisibleForTesting static final String HADOOP_CONFIG_FIELD = "hadoop_config";
47+
@VisibleForTesting static final String HADOOP_CONFIG_CAMEL_FIELD = "hadoopConfig";
48+
49+
static final String BEAM_WRITE_PROPERTY = "beam.write.";
50+
static final String BEAM_READ_PROPERTY = "beam.read.";
51+
52+
@VisibleForTesting final String tableLocation;
53+
@VisibleForTesting final @Nullable Long version;
54+
@VisibleForTesting final @Nullable String timestamp;
55+
@VisibleForTesting final @Nullable Map<String, String> hadoopConfig;
56+
57+
DeltaTable(Table table) {
58+
this(
59+
checkArgumentNotNull(
60+
table.getLocation(),
61+
"Delta Lake table location must be specified (catalog-based tables are not supported)."),
62+
table);
63+
}
64+
65+
DeltaTable(String tableLocation, Table table) {
66+
super(table.getSchema());
67+
this.schema = table.getSchema();
68+
this.tableLocation = tableLocation;
69+
70+
Long parsedVersion = null;
71+
String parsedTimestamp = null;
72+
Map<String, String> parsedHadoopConfig = new HashMap<>();
73+
74+
ObjectNode properties = table.getProperties();
75+
for (Map.Entry<String, JsonNode> property : properties.properties()) {
76+
String key = property.getKey();
77+
String lowerKey = key.toLowerCase();
78+
JsonNode val = property.getValue();
79+
80+
if (lowerKey.startsWith(BEAM_WRITE_PROPERTY)) {
81+
// TODO: Support writing to Delta Lake tables once a Delta Lake sink is
82+
// available.
83+
throw new IllegalArgumentException(
84+
String.format(
85+
"Beam write property '%s' is not supported. Writing to Delta Lake tables is currently not supported.",
86+
key));
87+
} else if (lowerKey.startsWith(BEAM_READ_PROPERTY)) {
88+
// none supported yet
89+
throw new IllegalArgumentException("Unknown Beam read property: " + key);
90+
} else if (lowerKey.equalsIgnoreCase(VERSION_FIELD)) {
91+
parsedVersion = parseVersion(val);
92+
} else if (lowerKey.equalsIgnoreCase(TIMESTAMP_FIELD)) {
93+
parsedTimestamp = val.asText();
94+
} else if (lowerKey.equalsIgnoreCase(HADOOP_CONFIG_FIELD)
95+
|| lowerKey.equalsIgnoreCase(HADOOP_CONFIG_CAMEL_FIELD)) {
96+
parseHadoopConfig(val, parsedHadoopConfig);
97+
} else {
98+
throw new IllegalArgumentException(String.format("Unknown property '%s'", key));
99+
}
100+
}
101+
102+
if (parsedVersion != null && parsedTimestamp != null) {
103+
throw new IllegalArgumentException("Cannot set both version and timestamp.");
104+
}
105+
106+
this.version = parsedVersion;
107+
this.timestamp = parsedTimestamp;
108+
this.hadoopConfig = parsedHadoopConfig.isEmpty() ? null : parsedHadoopConfig;
109+
}
110+
111+
private static Long parseVersion(JsonNode val) {
112+
if (val.isNumber()) {
113+
return val.asLong();
114+
}
115+
return Long.parseLong(val.asText());
116+
}
117+
118+
private static void parseHadoopConfig(JsonNode val, Map<String, String> targetMap) {
119+
if (val.isObject()) {
120+
Map<String, String> map =
121+
TableUtils.getObjectMapper()
122+
.convertValue(val, new TypeReference<Map<String, String>>() {});
123+
if (map != null) {
124+
targetMap.putAll(map);
125+
}
126+
} else if (val.isTextual()) {
127+
try {
128+
Map<String, String> map =
129+
TableUtils.getObjectMapper()
130+
.readValue(val.asText(), new TypeReference<Map<String, String>>() {});
131+
if (map != null) {
132+
targetMap.putAll(map);
133+
}
134+
} catch (Exception e) {
135+
throw new IllegalArgumentException("Failed to parse hadoop_config string as JSON", e);
136+
}
137+
}
138+
}
139+
140+
@Override
141+
public PCollection<Row> buildIOReader(PBegin begin) {
142+
return begin
143+
.apply(Managed.read(Managed.DELTA_LAKE).withConfig(getBaseConfig()))
144+
.getSinglePCollection();
145+
}
146+
147+
@Override
148+
public PCollection<Row> buildIOReader(
149+
PBegin begin, BeamSqlTableFilter filters, List<String> fieldNames) {
150+
// TODO: Support predicate pushdown and column pruning when supported by DeltaIO
151+
// / Managed Delta
152+
// Lake source.
153+
String error = "%s does not support predicate/project push-down, yet non-empty %s is passed.";
154+
if (!(filters instanceof DefaultTableFilter)) {
155+
throw new UnsupportedOperationException(
156+
String.format(error, this.getClass().getName(), "'filters'"));
157+
}
158+
if (!fieldNames.isEmpty()) {
159+
throw new UnsupportedOperationException(
160+
String.format(error, this.getClass().getName(), "'fieldNames'"));
161+
}
162+
return buildIOReader(begin);
163+
}
164+
165+
@Override
166+
public POutput buildIOWriter(PCollection<Row> input) {
167+
// TODO: Support writing to Delta Lake tables once a Delta Lake sink is
168+
// available.
169+
throw new UnsupportedOperationException(
170+
"Writing to Delta Lake tables is currently not supported.");
171+
}
172+
173+
@Override
174+
public PCollection.IsBounded isBounded() {
175+
return PCollection.IsBounded.BOUNDED;
176+
}
177+
178+
@Override
179+
public BeamSqlTableFilter constructFilter(List<RexNode> filter) {
180+
// TODO: Support predicate pushdown when supported by DeltaIO / Managed Delta
181+
// Lake source.
182+
return new DefaultTableFilter(filter);
183+
}
184+
185+
private Map<String, Object> getBaseConfig() {
186+
ImmutableMap.Builder<String, Object> managedConfigBuilder = ImmutableMap.builder();
187+
managedConfigBuilder.put("table", tableLocation);
188+
if (version != null) {
189+
managedConfigBuilder.put(VERSION_FIELD, version);
190+
}
191+
if (timestamp != null) {
192+
managedConfigBuilder.put(TIMESTAMP_FIELD, timestamp);
193+
}
194+
if (hadoopConfig != null && !hadoopConfig.isEmpty()) {
195+
managedConfigBuilder.put(HADOOP_CONFIG_FIELD, hadoopConfig);
196+
}
197+
return managedConfigBuilder.build();
198+
}
199+
}

‎sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProvider.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.sdk.extensions.sql.meta.provider.delta;
19+
20+
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
21+
22+
import com.google.auto.service.AutoService;
23+
import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
24+
import org.apache.beam.sdk.extensions.sql.meta.Table;
25+
import org.apache.beam.sdk.extensions.sql.meta.provider.InMemoryMetaTableProvider;
26+
import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
27+
28+
/**
29+
* {@link TableProvider} for Delta Lake tables.
30+
*
31+
* <p>A sample of Delta Lake table registration is:
32+
*
33+
* <pre>{@code
34+
* CREATE EXTERNAL TABLE orders(
35+
* id INTEGER,
36+
* name VARCHAR,
37+
* amount DOUBLE
38+
* )
39+
* TYPE 'delta'
40+
* LOCATION '/path/to/delta/orders'
41+
* TBLPROPERTIES '{"version": 1}'
42+
* }</pre>
43+
*/
44+
@AutoService(TableProvider.class)
45+
public class DeltaTableProvider extends InMemoryMetaTableProvider {
46+
47+
@Override
48+
public String getTableType() {
49+
return "delta";
50+
}
51+
52+
@Override
53+
public void createTable(Table table) {
54+
// TODO: Support catalog-based Delta Lake tables once Delta catalog support is implemented.
55+
checkArgument(
56+
table.getLocation() != null,
57+
"Delta Lake table location must be specified (catalog-based tables are not supported).");
58+
super.createTable(table);
59+
}
60+
61+
@Override
62+
public BeamSqlTable buildBeamSqlTable(Table table) {
63+
// TODO: Support catalog-based Delta Lake tables once Delta catalog support is implemented.
64+
checkArgument(
65+
table.getLocation() != null,
66+
"Delta Lake table location must be specified (catalog-based tables are not supported).");
67+
return new DeltaTable(table);
68+
}
69+
70+
@Override
71+
public boolean supportsPartitioning(Table table) {
72+
// TODO: Support partitioning when Delta Lake connector supports partitioned reads.
73+
return false;
74+
}
75+
}

‎sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/package-info.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
/** Table schema for Delta Lake. */
20+
package org.apache.beam.sdk.extensions.sql.meta.provider.delta;

0 commit comments

Comments
 (0)