You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/12/11 01:19:45 UTC
[incubator-iceberg] branch spark-3 updated (99edad1 -> 6f42a9f)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a change to branch spark-3
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git.
from 99edad1 Add friendly names to catalog tables (#421)
add 17effe1 Add case insensitive support to CachingCatalog (#671)
add d70a631 Fix MissingOverride error (#677)
add 6233d7a Add quick build option to disable checkstyle and error-prone (#685)
add 30fe730 Add compression level for Parquet (#689)
add dcc8866 Add Spark Avro JMH tests (#694)
add aad9f09 Remove some of the unused variables (#676)
new 1112066 Supports spark-3.0 data source V2 APIs
new 176393c Code review updates.
new 6f42a9f Add a SparkCatalog implementation.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/iceberg/catalog/TableIdentifier.java | 8 +
.../iceberg/expressions/BoundSetPredicate.java | 1 +
.../expressions/InclusiveMetricsEvaluator.java | 7 +-
.../iceberg/catalog/TestTableIdentifier.java | 13 +
baseline.gradle | 40 +--
build.gradle | 10 +-
.../iceberg/BaseMetastoreTableOperations.java | 1 -
.../java/org/apache/iceberg/CachingCatalog.java | 28 ++-
.../java/org/apache/iceberg/TableProperties.java | 3 +
.../org/apache/iceberg/hadoop/HadoopCatalog.java | 18 +-
.../java/org/apache/iceberg/hive/ClientPool.java | 2 -
.../java/org/apache/iceberg/hive/HiveCatalog.java | 19 +-
jmh.gradle | 4 +-
.../java/org/apache/iceberg/parquet/Parquet.java | 20 ++
.../IcebergSourceFlatAvroDataReadBenchmark.java} | 73 ++----
.../IcebergSourceNestedAvroDataReadBenchmark.java} | 78 +++---
.../org/apache/iceberg/spark/SparkCatalog.java | 252 +++++++++++++++++++
.../org/apache/iceberg/spark/SparkFilters.java | 19 ++
.../java/org/apache/iceberg/spark/SparkUtil.java | 242 ++++++++++++++++++
.../iceberg/spark/data/SparkParquetWriters.java | 2 -
.../apache/iceberg/spark/source/IcebergSource.java | 159 ++----------
.../source/{Reader.java => SparkBatchScan.java} | 237 ++++++++----------
.../source/{Writer.java => SparkBatchWrite.java} | 71 +++---
.../iceberg/spark/source/SparkScanBuilder.java | 102 ++++++++
...reamingWriter.java => SparkStreamingWrite.java} | 40 ++-
.../apache/iceberg/spark/source/SparkTable.java | 122 ++++++++++
.../iceberg/spark/source/SparkWriteBuilder.java | 173 +++++++++++++
.../source/{Stats.java => StagedSparkTable.java} | 23 +-
.../org/apache/iceberg/spark/source/Stats.java | 2 +-
.../iceberg/spark/data/TestSparkDateTimes.java | 7 +-
.../iceberg/spark/source/TestDataFrameWrites.java | 5 +
.../iceberg/spark/source/TestFilteredScan.java | 188 +++++++-------
.../iceberg/spark/source/TestIcebergSource.java | 6 +-
.../iceberg/spark/source/TestParquetWrite.java | 10 +
.../spark/source/TestStructuredStreaming.java | 5 +-
versions.lock | 269 +++++++++++----------
versions.props | 3 +-
37 files changed, 1564 insertions(+), 698 deletions(-)
copy spark/src/jmh/java/org/apache/iceberg/spark/source/{parquet/IcebergSourceFlatParquetDataReadBenchmark.java => avro/IcebergSourceFlatAvroDataReadBenchmark.java} (58%)
copy spark/src/jmh/java/org/apache/iceberg/spark/source/{parquet/IcebergSourceNestedParquetDataReadBenchmark.java => avro/IcebergSourceNestedAvroDataReadBenchmark.java} (58%)
create mode 100644 spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
create mode 100644 spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
rename spark/src/main/java/org/apache/iceberg/spark/source/{Reader.java => SparkBatchScan.java} (74%)
rename spark/src/main/java/org/apache/iceberg/spark/source/{Writer.java => SparkBatchWrite.java} (89%)
create mode 100644 spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
rename spark/src/main/java/org/apache/iceberg/spark/source/{StreamingWriter.java => SparkStreamingWrite.java} (70%)
create mode 100644 spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
create mode 100644 spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java
copy spark/src/main/java/org/apache/iceberg/spark/source/{Stats.java => StagedSparkTable.java} (65%)
[incubator-iceberg] 02/03: Code review updates.
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch spark-3
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
commit 176393ca8d9d3461d1e697e54e87b8b89c12ffad
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Mon Dec 9 16:54:25 2019 -0800
Code review updates.
---
api/src/main/java/org/apache/iceberg/Table.java | 9 -
build.gradle | 6 -
.../org/apache/iceberg/spark/SparkFilters.java | 19 +
.../java/org/apache/iceberg/spark/SparkUtil.java | 242 +++++++++++
.../java/org/apache/iceberg/spark/SparkUtils.java | 92 ----
...cebergTableProvider.java => IcebergSource.java} | 62 ++-
.../apache/iceberg/spark/source/IcebergTable.java | 222 ----------
.../{IcebergBatchScan.java => SparkBatchScan.java} | 482 +++++++++------------
...cebergBatchWriter.java => SparkBatchWrite.java} | 77 ++--
.../iceberg/spark/source/SparkScanBuilder.java | 102 +++++
...reamingWriter.java => SparkStreamingWrite.java} | 54 +--
.../apache/iceberg/spark/source/SparkTable.java | 118 +++++
.../iceberg/spark/source/SparkWriteBuilder.java | 173 ++++++++
...org.apache.spark.sql.sources.DataSourceRegister | 2 +-
.../iceberg/spark/source/TestDataFrameWrites.java | 12 +-
.../iceberg/spark/source/TestFilteredScan.java | 64 +--
.../iceberg/spark/source/TestIcebergSource.java | 5 +-
.../iceberg/spark/source/TestParquetWrite.java | 21 +-
.../spark/source/TestStructuredStreaming.java | 4 +-
19 files changed, 991 insertions(+), 775 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/Table.java b/api/src/main/java/org/apache/iceberg/Table.java
index e0fa525..63c55e5 100644
--- a/api/src/main/java/org/apache/iceberg/Table.java
+++ b/api/src/main/java/org/apache/iceberg/Table.java
@@ -228,13 +228,4 @@ public interface Table {
* @return a {@link LocationProvider} to provide locations for new data files
*/
LocationProvider locationProvider();
-
- /**
- * Return the name of this table.
- *
- * @return this table's name
- */
- default String name() {
- return "table(" + hashCode() + ")";
- }
}
diff --git a/build.gradle b/build.gradle
index 6835cbe..cc0f666 100644
--- a/build.gradle
+++ b/build.gradle
@@ -240,15 +240,9 @@ project(':iceberg-spark') {
compile project(':iceberg-hive')
compileOnly "org.apache.avro:avro"
- compileOnly("org.apache.spark:spark-sql_2.12") {
- exclude group: 'org.apache.avro', module: 'avro'
- }
compileOnly("org.apache.spark:spark-hive_2.12") {
exclude group: 'org.apache.avro', module: 'avro'
}
- compileOnly("org.apache.spark:spark-catalyst_2.12") {
- exclude group: 'org.apache.avro', module: 'avro'
- }
testCompile "org.apache.hadoop:hadoop-hdfs::tests"
testCompile "org.apache.hadoop:hadoop-common::tests"
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkFilters.java b/spark/src/main/java/org/apache/iceberg/spark/SparkFilters.java
index c6f8f52..6c43b87 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/SparkFilters.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkFilters.java
@@ -25,7 +25,10 @@ import java.sql.Date;
import java.sql.Timestamp;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.spark.sql.catalyst.util.DateTimeUtils;
+import org.apache.spark.sql.sources.AlwaysFalse$;
+import org.apache.spark.sql.sources.AlwaysTrue$;
import org.apache.spark.sql.sources.And;
import org.apache.spark.sql.sources.EqualNullSafe;
import org.apache.spark.sql.sources.EqualTo;
@@ -60,6 +63,8 @@ public class SparkFilters {
private static final ImmutableMap<Class<? extends Filter>, Operation> FILTERS = ImmutableMap
.<Class<? extends Filter>, Operation>builder()
+ .put(AlwaysTrue$.class, Operation.TRUE)
+ .put(AlwaysFalse$.class, Operation.FALSE)
.put(EqualTo.class, Operation.EQ)
.put(EqualNullSafe.class, Operation.EQ)
.put(GreaterThan.class, Operation.GT)
@@ -75,11 +80,25 @@ public class SparkFilters {
.put(StringStartsWith.class, Operation.STARTS_WITH)
.build();
+ public static Expression convert(Filter[] filters) {
+ Expression expression = Expressions.alwaysTrue();
+ for (Filter filter : filters) {
+ expression = Expressions.and(expression, convert(filter));
+ }
+ return expression;
+ }
+
public static Expression convert(Filter filter) {
// avoid using a chain of if instanceof statements by mapping to the expression enum.
Operation op = FILTERS.get(filter.getClass());
if (op != null) {
switch (op) {
+ case TRUE:
+ return Expressions.alwaysTrue();
+
+ case FALSE:
+ return Expressions.alwaysFalse();
+
case IS_NULL:
IsNull isNullFilter = (IsNull) filter;
return isNull(isNullFilter.attribute());
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java b/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
new file mode 100644
index 0000000..62b83f2
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.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.iceberg.spark;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.transforms.PartitionSpecVisitor;
+import org.apache.iceberg.types.Type;
+import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.connector.expressions.Expression;
+import org.apache.spark.sql.connector.expressions.Expressions;
+import org.apache.spark.sql.connector.expressions.Literal;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.sql.types.LongType;
+
+public class SparkUtil {
+
+ private SparkUtil() {
+ }
+
+ private static final Joiner DOT = Joiner.on(".");
+
+ /**
+ * Applies a list of Spark table changes to an {@link UpdateProperties} operation.
+ * <p>
+ * All non-property changes in the list are ignored.
+ *
+ * @param pendingUpdate an uncommitted UpdateProperties operation to configure
+ * @param changes a list of Spark table changes
+ * @return the UpdateProperties operation configured with the changes
+ */
+ public static UpdateProperties applyPropertyChanges(UpdateProperties pendingUpdate, List<TableChange> changes) {
+ for (TableChange change : changes) {
+ if (change instanceof TableChange.SetProperty) {
+ TableChange.SetProperty set = (TableChange.SetProperty) change;
+ pendingUpdate.set(set.property(), set.value());
+
+ } else if (change instanceof TableChange.RemoveProperty) {
+ TableChange.RemoveProperty remove = (TableChange.RemoveProperty) change;
+ pendingUpdate.remove(remove.property());
+ }
+ }
+
+ return pendingUpdate;
+ }
+
+ /**
+ * Applies a list of Spark table changes to an {@link UpdateSchema} operation.
+ * <p>
+ * All non-schema changes in the list are ignored.
+ *
+ * @param pendingUpdate an uncommitted UpdateSchema operation to configure
+ * @param changes a list of Spark table changes
+ * @return the UpdateSchema operation configured with the changes
+ */
+ public static UpdateSchema applySchemaChanges(UpdateSchema pendingUpdate, List<TableChange> changes) {
+ for (TableChange change : changes) {
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ Type type = SparkSchemaUtil.convert(add.dataType());
+ pendingUpdate.addColumn(parentName(add.fieldNames()), leafName(add.fieldNames()), type, add.comment());
+
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ TableChange.UpdateColumnType update = (TableChange.UpdateColumnType) change;
+ Type newType = SparkSchemaUtil.convert(update.newDataType());
+ Preconditions.checkArgument(newType.isPrimitiveType(),
+ "Cannot update '%s', not a primitive type: %s", DOT.join(update.fieldNames()), update.newDataType());
+ pendingUpdate.updateColumn(DOT.join(update.fieldNames()), newType.asPrimitiveType());
+
+ } else if (change instanceof TableChange.UpdateColumnComment) {
+ TableChange.UpdateColumnComment update = (TableChange.UpdateColumnComment) change;
+ pendingUpdate.updateColumnDoc(DOT.join(update.fieldNames()), update.newComment());
+
+ } else if (change instanceof TableChange.RenameColumn) {
+ TableChange.RenameColumn rename = (TableChange.RenameColumn) change;
+ pendingUpdate.renameColumn(DOT.join(rename.fieldNames()), rename.newName());
+
+ } else if (change instanceof TableChange.DeleteColumn) {
+ TableChange.DeleteColumn delete = (TableChange.DeleteColumn) change;
+ pendingUpdate.deleteColumn(DOT.join(delete.fieldNames()));
+ }
+ }
+
+ return pendingUpdate;
+ }
+
+ /**
+ * Converts a PartitionSpec to Spark transforms.
+ *
+ * @param spec a PartitionSpec
+ * @return an array of Transforms
+ */
+ public static Transform[] toTransforms(PartitionSpec spec) {
+ List<Transform> transforms = PartitionSpecVisitor.visit(spec.schema(), spec,
+ new PartitionSpecVisitor<Transform>() {
+ @Override
+ public Transform identity(String sourceName, int sourceId) {
+ return Expressions.identity(sourceName);
+ }
+
+ @Override
+ public Transform bucket(String sourceName, int sourceId, int width) {
+ return Expressions.bucket(width, sourceName);
+ }
+
+ @Override
+ public Transform truncate(String sourceName, int sourceId, int width) {
+ return Expressions.apply("truncate", Expressions.column(sourceName), Expressions.literal(width));
+ }
+
+ @Override
+ public Transform year(String sourceName, int sourceId) {
+ return Expressions.years(sourceName);
+ }
+
+ @Override
+ public Transform month(String sourceName, int sourceId) {
+ return Expressions.months(sourceName);
+ }
+
+ @Override
+ public Transform day(String sourceName, int sourceId) {
+ return Expressions.days(sourceName);
+ }
+
+ @Override
+ public Transform hour(String sourceName, int sourceId) {
+ return Expressions.hours(sourceName);
+ }
+ });
+
+ return transforms.toArray(new Transform[0]);
+ }
+
+ /**
+ * Converts Spark transforms into a {@link PartitionSpec}.
+ *
+ * @param schema the table schema
+ * @param partitioning Spark Transforms
+ * @return a PartitionSpec
+ */
+ public static PartitionSpec toPartitionSpec(Schema schema, Transform[] partitioning) {
+ if (partitioning == null || partitioning.length == 0) {
+ return PartitionSpec.unpartitioned();
+ }
+
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(schema);
+ for (Transform transform : partitioning) {
+ Preconditions.checkArgument(transform.references().length == 1,
+ "Cannot convert transform with more than one column reference: %s", transform);
+ String colName = DOT.join(transform.references()[0].fieldNames());
+ switch (transform.name()) {
+ case "identity":
+ builder.identity(colName);
+ break;
+ case "bucket":
+ builder.bucket(colName, findWidth(transform));
+ break;
+ case "years":
+ builder.year(colName);
+ break;
+ case "months":
+ builder.month(colName);
+ break;
+ case "days":
+ builder.day(colName);
+ break;
+ case "hours":
+ builder.hour(colName);
+ break;
+ case "truncate":
+ builder.truncate(colName, findWidth(transform));
+ break;
+ default:
+ throw new UnsupportedOperationException("Transform is not supported: " + transform);
+ }
+ }
+
+ return builder.build();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static int findWidth(Transform transform) {
+ for (Expression expr : transform.arguments()) {
+ if (expr instanceof Literal) {
+ if (((Literal) expr).dataType() instanceof IntegerType) {
+ Literal<Integer> lit = (Literal<Integer>) expr;
+ Preconditions.checkArgument(lit.value() > 0,
+ "Unsupported width for transform: %s", transform.describe());
+ return lit.value();
+
+ } else if (((Literal) expr).dataType() instanceof LongType) {
+ Literal<Long> lit = (Literal<Long>) expr;
+ Preconditions.checkArgument(lit.value() > 0 && lit.value() < Integer.MAX_VALUE,
+ "Unsupported width for transform: %s", transform.describe());
+ if (lit.value() > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException();
+ }
+ return lit.value().intValue();
+ }
+ }
+ }
+
+ throw new IllegalArgumentException("Cannot find width for transform: " + transform.describe());
+ }
+
+ private static String leafName(String[] fieldNames) {
+ Preconditions.checkArgument(fieldNames.length > 0, "Invalid field name: at least one name is required");
+ return fieldNames[fieldNames.length - 1];
+ }
+
+ private static String parentName(String[] fieldNames) {
+ if (fieldNames.length > 1) {
+ return DOT.join(Arrays.copyOfRange(fieldNames, 0, fieldNames.length - 1));
+ }
+ return null;
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java b/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java
deleted file mode 100644
index 64e81cb..0000000
--- a/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.iceberg.spark;
-
-
-import com.google.common.collect.Lists;
-import java.util.List;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.connector.expressions.Expressions;
-import org.apache.spark.sql.connector.expressions.Transform;
-
-public final class SparkUtils {
-
- private static final Pattern HAS_WIDTH = Pattern.compile("(\\w+)\\[(\\d+)\\]");
-
- private SparkUtils() {}
-
- public static SparkSession getSparkSession() {
- return SparkSession.builder().getOrCreate();
- }
-
- public static Configuration getBaseConf() {
- return getSparkSession().sparkContext().hadoopConfiguration();
- }
-
-
- public static Transform[] toTransforms(PartitionSpec spec) {
- List<Transform> transforms = Lists.newArrayList();
- int numBuckets = 0;
- List<String> bucketColumns = Lists.newArrayList();
-
- for (PartitionField f : spec.fields()) {
- Matcher widthMatcher = HAS_WIDTH.matcher(f.transform().toString());
- if (widthMatcher.matches()) {
- String name = widthMatcher.group(1);
- if (name.equalsIgnoreCase("truncate")) {
- throw new UnsupportedOperationException("Spark doesn't support truncate transform");
-
- } else if (name.equalsIgnoreCase("bucket")) {
- numBuckets = Integer.parseInt(widthMatcher.group(2));
- bucketColumns.add(spec.schema().findColumnName(f.sourceId()));
-
- } else if (f.transform().toString().equalsIgnoreCase("identity")) {
- transforms.add(Expressions.identity(spec.schema().findColumnName(f.sourceId())));
-
- } else if (f.transform().toString().equalsIgnoreCase("year")) {
- transforms.add(Expressions.years(spec.schema().findColumnName(f.sourceId())));
-
- } else if (f.transform().toString().equalsIgnoreCase("month")) {
- transforms.add(Expressions.months(spec.schema().findColumnName(f.sourceId())));
-
- } else if (f.transform().toString().equalsIgnoreCase("day")) {
- transforms.add(Expressions.days(spec.schema().findColumnName(f.sourceId())));
-
- } else if (f.transform().toString().equalsIgnoreCase("hour")) {
- transforms.add(Expressions.hours(spec.schema().findColumnName(f.sourceId())));
-
- } else {
- throw new UnsupportedOperationException("Spark doesn't support transform " + f.transform());
- }
- }
- }
-
- if (!bucketColumns.isEmpty()) {
- transforms.add(Expressions.bucket(numBuckets, bucketColumns.toArray(new String[0])));
- }
-
- return transforms.toArray(new Transform[0]);
- }
-}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
similarity index 60%
rename from spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
index 9f7e1d9..0b1cb7a 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
@@ -21,47 +21,59 @@ package org.apache.iceberg.spark.source;
import com.google.common.base.Preconditions;
import java.util.Map;
-import java.util.Optional;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.hive.HiveCatalogs;
-import org.apache.iceberg.spark.SparkUtils;
-import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.connector.catalog.TableProvider;
import org.apache.spark.sql.sources.DataSourceRegister;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-public class IcebergTableProvider implements DataSourceRegister, TableProvider {
+public class IcebergSource implements DataSourceRegister, TableProvider {
@Override
public String shortName() {
return "iceberg";
}
@Override
- public Table getTable(CaseInsensitiveStringMap options) {
+ public SparkTable getTable(CaseInsensitiveStringMap options) {
return getTable(options, null);
}
@Override
- public Table getTable(CaseInsensitiveStringMap options, StructType readSchema) {
+ public SparkTable getTable(CaseInsensitiveStringMap options, StructType readSchema) {
// Get Iceberg table from options
- Configuration conf = new Configuration(SparkUtils.getBaseConf());
- org.apache.iceberg.Table tableInIceberg = getTableAndResolveHadoopConfiguration(options, conf);
+ Configuration conf = new Configuration(SparkSession.active().sparkContext().hadoopConfiguration());
+ Table icebergTable = getTableAndResolveHadoopConfiguration(options, conf);
// Build Spark table based on Iceberg table, and return it
- return new IcebergTable(tableInIceberg, readSchema);
+ return new SparkTable(icebergTable, readSchema);
}
- protected org.apache.iceberg.Table getTableAndResolveHadoopConfiguration(
+ protected Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
+ Preconditions.checkArgument(options.containsKey("path"), "Cannot open table: path is not set");
+ String path = options.get("path");
+
+ if (path.contains("/")) {
+ HadoopTables tables = new HadoopTables(conf);
+ return tables.load(path);
+ } else {
+ HiveCatalog hiveCatalog = HiveCatalogs.loadCatalog(conf);
+ TableIdentifier tableIdentifier = TableIdentifier.parse(path);
+ return hiveCatalog.loadTable(tableIdentifier);
+ }
+ }
+
+ private Table getTableAndResolveHadoopConfiguration(
CaseInsensitiveStringMap options, Configuration conf) {
// Overwrite configurations from the Spark Context with configurations from the options.
mergeIcebergHadoopConfs(conf, options);
- // Find table (in Iceberg) based on the given path
- org.apache.iceberg.Table table = findTable(options, conf);
+ Table table = findTable(options, conf);
// Set confs from table properties
mergeIcebergHadoopConfs(conf, table.properties());
@@ -72,30 +84,10 @@ public class IcebergTableProvider implements DataSourceRegister, TableProvider {
return table;
}
- /**
- * Merge delta options into base conf
- *
- * @param baseConf the base conf
- * @param options the delta options to merge into base
- */
- private void mergeIcebergHadoopConfs(Configuration baseConf, Map<String, String> options) {
+ private static void mergeIcebergHadoopConfs(
+ Configuration baseConf, Map<String, String> options) {
options.keySet().stream()
- .filter(key -> key.startsWith("hadoop.")) /* filter all keys staring with "hadoop." */
+ .filter(key -> key.startsWith("hadoop."))
.forEach(key -> baseConf.set(key.replaceFirst("hadoop.", ""), options.get(key)));
- /* Modify the key by removing the prefix of "hadoop." and merge into base */
- }
-
- protected org.apache.iceberg.Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
- Optional<String> path = Optional.ofNullable(options.get("path"));
- Preconditions.checkArgument(path.isPresent(), "Cannot open table: path is not set");
-
- if (path.get().contains("/")) { // hadoop table
- HadoopTables tables = new HadoopTables(conf);
- return tables.load(path.get());
- } else { // hive table
- HiveCatalog hiveCatalog = HiveCatalogs.loadCatalog(conf);
- TableIdentifier tableIdentifier = TableIdentifier.parse(path.get());
- return hiveCatalog.loadTable(tableIdentifier);
- }
}
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java
deleted file mode 100644
index d02645d..0000000
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java
+++ /dev/null
@@ -1,222 +0,0 @@
-/*
- * 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.iceberg.spark.source;
-
-import com.google.common.collect.ImmutableSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.spark.SparkSchemaUtil;
-import org.apache.iceberg.spark.SparkUtils;
-import org.apache.iceberg.transforms.UnknownTransform;
-import org.apache.iceberg.types.CheckCompatibility;
-import org.apache.spark.sql.connector.catalog.SupportsRead;
-import org.apache.spark.sql.connector.catalog.SupportsWrite;
-import org.apache.spark.sql.connector.catalog.Table;
-import org.apache.spark.sql.connector.catalog.TableCapability;
-import org.apache.spark.sql.connector.expressions.Transform;
-import org.apache.spark.sql.connector.read.ScanBuilder;
-import org.apache.spark.sql.connector.write.BatchWrite;
-import org.apache.spark.sql.connector.write.SupportsDynamicOverwrite;
-import org.apache.spark.sql.connector.write.SupportsTruncate;
-import org.apache.spark.sql.connector.write.WriteBuilder;
-import org.apache.spark.sql.connector.write.streaming.StreamingWrite;
-import org.apache.spark.sql.types.StructType;
-import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-
-final class IcebergTable implements Table, SupportsRead, SupportsWrite {
-
- private static final Set<TableCapability> CAPABILITIES = ImmutableSet.of(
- TableCapability.BATCH_READ,
- TableCapability.BATCH_WRITE,
- TableCapability.MICRO_BATCH_READ,
- TableCapability.STREAMING_WRITE,
- TableCapability.TRUNCATE,
- TableCapability.OVERWRITE_DYNAMIC);
-
- private final org.apache.iceberg.Table tableInIceberg;
- private StructType requestSchema;
-
- IcebergTable(org.apache.iceberg.Table tableInIceberg, StructType requestSchema) {
- this.tableInIceberg = tableInIceberg;
-
- if (requestSchema != null) {
- SparkSchemaUtil.convert(tableInIceberg.schema(), requestSchema);
- this.requestSchema = requestSchema;
- }
- }
-
- @Override
- public String name() {
- return tableInIceberg.name();
- }
-
- @Override
- public StructType schema() {
- if (requestSchema != null) {
- return requestSchema;
- }
- return SparkSchemaUtil.convert(tableInIceberg.schema());
- }
-
- @Override
- public Transform[] partitioning() {
- return SparkUtils.toTransforms(tableInIceberg.spec());
- }
-
- @Override
- public Map<String, String> properties() {
- return tableInIceberg.properties();
- }
-
- @Override
- public Set<TableCapability> capabilities() {
- return CAPABILITIES;
- }
-
- @Override
- public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
- return () -> new IcebergBatchScan(tableInIceberg, options, requestSchema);
- }
-
- @Override
- public WriteBuilder newWriteBuilder(CaseInsensitiveStringMap options) {
- return new IcebergWriteBuilder(tableInIceberg, options);
- }
-
- static class IcebergWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsTruncate {
-
- private org.apache.iceberg.Table table;
- private CaseInsensitiveStringMap writeOptions;
- private TableCapability writeBehavior = TableCapability.BATCH_WRITE;
- private String writeQueryId = null;
- private StructType dsStruct = null;
-
- IcebergWriteBuilder(org.apache.iceberg.Table table, CaseInsensitiveStringMap options) {
- this.table = table;
- this.writeOptions = options;
- }
-
- @Override
- public WriteBuilder withQueryId(String queryId) {
- this.writeQueryId = queryId;
- return this;
- }
-
- @Override
- public WriteBuilder withInputDataSchema(StructType schemaInput) {
- this.dsStruct = schemaInput;
- return this;
- }
-
- @Override
- public WriteBuilder overwriteDynamicPartitions() {
- this.writeBehavior = TableCapability.OVERWRITE_DYNAMIC;
- return this;
- }
-
- @Override
- public WriteBuilder truncate() {
- this.writeBehavior = TableCapability.TRUNCATE;
- return this;
- }
-
- @Override
- public BatchWrite buildForBatch() {
- // TODO. Check queryId and schema before build?
-
- // Validate
- Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
- validateWriteSchema(table.schema(), dsSchema, checkNullability(writeOptions));
- validatePartitionTransforms(table.spec());
-
- // Get application id
- String appId = SparkUtils.getSparkSession().sparkContext().applicationId();
-
- // Get write-audit-publish id
- String wapId = SparkUtils.getSparkSession().conf().get("spark.wap.id", null);
-
- return new IcebergBatchWriter(table, writeOptions, writeBehavior, appId, wapId, dsSchema);
- }
-
- @Override
- public StreamingWrite buildForStreaming() {
- // TODO. Check queryId and schema before build?
-
- // Validate
- Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
- validateWriteSchema(table.schema(), dsSchema, checkNullability(writeOptions));
- validatePartitionTransforms(table.spec());
-
- // Change to streaming write if it is just append
- if (writeBehavior.equals(TableCapability.BATCH_WRITE)) {
- writeBehavior = TableCapability.STREAMING_WRITE;
- }
-
- // Get application id
- String appId = SparkUtils.getSparkSession().sparkContext().applicationId();
- String wapId = SparkUtils.getSparkSession().conf().get("spark.wap.id", null);
- return new IcebergStreamingWriter(table, writeOptions, writeQueryId, writeBehavior, appId, wapId, table.schema());
- }
-
- private void validateWriteSchema(Schema tableSchema, Schema dsSchema, Boolean checkNullability) {
- List<String> errors;
- if (checkNullability) {
- errors = CheckCompatibility.writeCompatibilityErrors(tableSchema, dsSchema);
- } else {
- errors = CheckCompatibility.typeCompatibilityErrors(tableSchema, dsSchema);
- }
- if (!errors.isEmpty()) {
- StringBuilder sb = new StringBuilder();
- sb.append("Cannot write incompatible dataset to table with schema:\n")
- .append(tableSchema)
- .append("\nProblems:");
- for (String error : errors) {
- sb.append("\n* ").append(error);
- }
- throw new IllegalArgumentException(sb.toString());
- }
- }
-
- private void validatePartitionTransforms(PartitionSpec spec) {
- if (spec.fields().stream().anyMatch(field -> field.transform() instanceof UnknownTransform)) {
- String unsupported = spec.fields().stream()
- .map(PartitionField::transform)
- .filter(transform -> transform instanceof UnknownTransform)
- .map(org.apache.iceberg.transforms.Transform::toString)
- .collect(Collectors.joining(", "));
-
- throw new UnsupportedOperationException(
- String.format("Cannot write using unsupported transforms: %s", unsupported));
- }
- }
-
- private boolean checkNullability(CaseInsensitiveStringMap options) {
- boolean sparkCheckNullability = Boolean.parseBoolean(SparkUtils.getSparkSession().conf()
- .get("spark.sql.iceberg.check-nullability", "true"));
- boolean dataFrameCheckNullability = options.getBoolean("check-nullability", true);
- return sparkCheckNullability && dataFrameCheckNullability;
- }
- }
-}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java
similarity index 77%
rename from spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java
index b3fef6d..4e04749 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java
@@ -28,6 +28,7 @@ import java.io.Closeable;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -56,7 +57,6 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.orc.ORC;
import org.apache.iceberg.parquet.Parquet;
-import org.apache.iceberg.spark.SparkFilters;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.spark.data.SparkAvroReader;
import org.apache.iceberg.spark.data.SparkOrcReader;
@@ -76,10 +76,7 @@ import org.apache.spark.sql.connector.read.PartitionReader;
import org.apache.spark.sql.connector.read.PartitionReaderFactory;
import org.apache.spark.sql.connector.read.Scan;
import org.apache.spark.sql.connector.read.Statistics;
-import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
-import org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns;
import org.apache.spark.sql.connector.read.SupportsReportStatistics;
-import org.apache.spark.sql.sources.Filter;
import org.apache.spark.sql.types.BinaryType;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.Decimal;
@@ -91,14 +88,12 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.apache.spark.unsafe.types.UTF8String;
import scala.collection.JavaConverters;
-public class IcebergBatchScan implements Scan,
- Batch,
- SupportsPushDownFilters,
- SupportsPushDownRequiredColumns,
- SupportsReportStatistics {
- private static final Filter[] NO_FILTERS = new Filter[0];
+class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
private final Table table;
+ private final boolean caseSensitive;
+ private final Schema expectedSchema;
+ private final List<Expression> filterExpressions;
private final Long snapshotId;
private final Long asOfTimestamp;
private final Long splitSize;
@@ -106,18 +101,16 @@ public class IcebergBatchScan implements Scan,
private final Long splitOpenFileCost;
private final FileIO fileIo;
private final EncryptionManager encryptionManager;
- private final boolean caseSensitive;
- private StructType requestedSchema = null;
- private List<Expression> filterExpressions = null;
- private Filter[] pushedFilters = NO_FILTERS;
// lazy variables
- private Schema schema = null;
- private StructType type = null; // cached because Spark accesses it multiple times
private List<CombinedScanTask> tasks = null; // lazy cache of tasks
- public IcebergBatchScan(Table table, Boolean caseSensitive, CaseInsensitiveStringMap options) {
+ SparkBatchScan(Table table, boolean caseSensitive, Schema expectedSchema, List<Expression> filters,
+ CaseInsensitiveStringMap options) {
this.table = table;
+ this.caseSensitive = caseSensitive;
+ this.expectedSchema = expectedSchema;
+ this.filterExpressions = filters;
this.snapshotId = options.containsKey("snapshot-id") ? options.getLong("snapshot-id", 0) : null;
this.asOfTimestamp = options.containsKey("as-of-timestamp") ? options.getLong("as-of-timestamp", 0) : null;
@@ -134,22 +127,53 @@ public class IcebergBatchScan implements Scan,
this.splitOpenFileCost = options.containsKey("file-open-cost") ? options.getLong("file-open-cost",
TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT) : null;
- this.schema = table.schema();
this.fileIo = table.io();
this.encryptionManager = table.encryption();
- this.caseSensitive = caseSensitive;
}
- public IcebergBatchScan(Table table, CaseInsensitiveStringMap options) {
- this(table, true, options);
+ @Override
+ public Batch toBatch() {
+ return this;
}
- public IcebergBatchScan(Table table, CaseInsensitiveStringMap options, StructType requestedSchema) {
- this(table, true, options);
+ @Override
+ public StructType readSchema() {
+ return SparkSchemaUtil.convert(expectedSchema);
+ }
- if (requestedSchema != null) {
- pruneColumns(requestedSchema);
+ @Override
+ public InputPartition[] planInputPartitions() {
+ String tableSchemaString = SchemaParser.toJson(table.schema());
+ String expectedSchemaString = SchemaParser.toJson(expectedSchema);
+
+ List<CombinedScanTask> scanTasks = tasks();
+ InputPartition[] readTasks = new InputPartition[scanTasks.size()];
+ for (int i = 0; i < scanTasks.size(); i++) {
+ readTasks[i] = new ReadTask(
+ scanTasks.get(i), tableSchemaString, expectedSchemaString, fileIo, encryptionManager, caseSensitive);
}
+
+ return readTasks;
+ }
+
+ @Override
+ public ReaderFactory createReaderFactory() {
+ return new ReaderFactory();
+ }
+
+ @Override
+ public Statistics estimateStatistics() {
+ long sizeInBytes = 0L;
+ long numRows = 0L;
+
+ for (CombinedScanTask task : tasks()) {
+ for (FileScanTask file : task.files()) {
+ sizeInBytes += file.length();
+ numRows += file.file().recordCount();
+ }
+ }
+
+ return new Stats(sizeInBytes, numRows);
}
private List<CombinedScanTask> tasks() {
@@ -157,7 +181,7 @@ public class IcebergBatchScan implements Scan,
TableScan scan = table
.newScan()
.caseSensitive(caseSensitive)
- .project(lazySchema());
+ .project(expectedSchema);
if (snapshotId != null) {
scan = scan.useSnapshot(snapshotId);
@@ -196,301 +220,96 @@ public class IcebergBatchScan implements Scan,
}
@Override
- public Filter[] pushFilters(Filter[] filters) {
- this.tasks = null; // invalidate cached tasks, if present
-
- List<Expression> expressions = Lists.newArrayListWithExpectedSize(filters.length);
- List<Filter> pushed = Lists.newArrayListWithExpectedSize(filters.length);
-
- for (Filter filter : filters) {
- Expression expr = SparkFilters.convert(filter);
- if (expr != null) {
- expressions.add(expr);
- pushed.add(filter);
- }
- }
-
- this.filterExpressions = expressions;
- this.pushedFilters = pushed.toArray(new Filter[0]);
-
- // invalidate the schema that will be projected
- this.schema = null;
- this.type = null;
-
- // Spark doesn't support residuals per task, so return all filters
- // to get Spark to handle record-level filtering
- return filters;
- }
-
- @Override
- public Batch toBatch() {
- return this;
- }
-
- @Override
- public Filter[] pushedFilters() {
- return pushedFilters;
- }
-
- @Override
- public void pruneColumns(StructType newRequestedSchema) {
- this.requestedSchema = newRequestedSchema;
-
- // invalidate the schema that will be projected
- this.schema = null;
- this.type = null;
+ public String toString() {
+ return String.format(
+ "IcebergScan(table=%s, type=%s, filters=%s, caseSensitive=%s)",
+ table, expectedSchema.asStruct(), filterExpressions, caseSensitive);
}
- @Override
- public Scan build() {
- return this;
- }
-
- @Override
- public Statistics estimateStatistics() {
- long sizeInBytes = 0L;
- long numRows = 0L;
-
- for (CombinedScanTask task : tasks()) {
- for (FileScanTask file : task.files()) {
- sizeInBytes += file.length();
- numRows += file.file().recordCount();
+ private static class ReaderFactory implements PartitionReaderFactory {
+ @Override
+ public PartitionReader<InternalRow> createReader(InputPartition inputPartition) {
+ if (inputPartition instanceof ReadTask) {
+ return new TaskDataReader((ReadTask) inputPartition);
+ } else {
+ throw new UnsupportedOperationException("Incorrect input partition type: " + inputPartition);
}
}
-
- return new Stats(sizeInBytes, numRows);
}
- public static class BatchReadInputPartition implements InputPartition, Serializable {
+ private static class ReadTask implements InputPartition, Serializable {
private final CombinedScanTask task;
private final String tableSchemaString;
private final String expectedSchemaString;
- private final FileIO fileIo;
+ private final FileIO io;
private final EncryptionManager encryptionManager;
private final boolean caseSensitive;
private transient Schema tableSchema = null;
private transient Schema expectedSchema = null;
- BatchReadInputPartition(
- CombinedScanTask task,
- String tableSchemaString,
- String expectedSchemaString,
- FileIO fileIo,
- EncryptionManager encryptionManager,
- boolean caseSensitive) {
+ ReadTask(CombinedScanTask task, String tableSchemaString, String expectedSchemaString,
+ FileIO io, EncryptionManager encryptionManager, boolean caseSensitive) {
this.task = task;
this.tableSchemaString = tableSchemaString;
this.expectedSchemaString = expectedSchemaString;
- this.fileIo = fileIo;
+ this.io = io;
this.encryptionManager = encryptionManager;
this.caseSensitive = caseSensitive;
}
- private Schema lazyTableSchema() {
- if (tableSchema == null) {
- this.tableSchema = SchemaParser.fromJson(tableSchemaString);
- }
- return tableSchema;
+ public Collection<FileScanTask> files() {
+ return task.files();
}
- private Schema lazyExpectedSchema() {
- if (expectedSchema == null) {
- this.expectedSchema = SchemaParser.fromJson(expectedSchemaString);
- }
- return expectedSchema;
- }
- }
-
- @Override
- public InputPartition[] planInputPartitions() {
- String tableSchemaString = SchemaParser.toJson(table.schema());
- String expectedSchemaString = SchemaParser.toJson(lazySchema());
-
- List<CombinedScanTask> scanTasks = tasks();
- InputPartition[] readTasks = new InputPartition[scanTasks.size()];
- for (int i = 0; i < scanTasks.size(); i++) {
- readTasks[i] = new BatchReadInputPartition(scanTasks.get(i), tableSchemaString, expectedSchemaString, fileIo,
- encryptionManager, caseSensitive);
- }
-
- return readTasks;
- }
-
- @Override
- public IcebergRowReaderFactory createReaderFactory() {
- return new IcebergRowReaderFactory();
- }
-
- private Schema lazySchema() {
- if (schema == null) {
- if (requestedSchema != null) {
- this.schema = SparkSchemaUtil.prune(table.schema(), requestedSchema);
- } else {
- this.schema = table.schema();
- }
+ public FileIO io() {
+ return io;
}
- return schema;
- }
-
- private StructType lazyType() {
- if (type == null) {
- this.type = SparkSchemaUtil.convert(lazySchema());
- }
- return type;
- }
-
- @Override
- public StructType readSchema() {
- return lazyType();
- }
-
-
- public static class PartitionRowConverter implements Function<StructLike, InternalRow> {
- private final DataType[] types;
- private final int[] positions;
- private final Class<?>[] javaTypes;
- private final GenericInternalRow reusedRow;
-
- PartitionRowConverter(Schema partitionSchema, PartitionSpec spec) {
- StructType partitionType = SparkSchemaUtil.convert(partitionSchema);
- StructField[] fields = partitionType.fields();
-
- this.types = new DataType[fields.length];
- this.positions = new int[types.length];
- this.javaTypes = new Class<?>[types.length];
- this.reusedRow = new GenericInternalRow(types.length);
-
- List<PartitionField> partitionFields = spec.fields();
- for (int rowIndex = 0; rowIndex < fields.length; rowIndex += 1) {
- this.types[rowIndex] = fields[rowIndex].dataType();
- int sourceId = partitionSchema.columns().get(rowIndex).fieldId();
- for (int specIndex = 0; specIndex < partitionFields.size(); specIndex += 1) {
- PartitionField field = spec.fields().get(specIndex);
- if (field.sourceId() == sourceId && "identity".equals(field.transform().toString())) {
- positions[rowIndex] = specIndex;
- javaTypes[rowIndex] = spec.javaClasses()[specIndex];
- break;
- }
- }
- }
+ public EncryptionManager encryptionManager() {
+ return encryptionManager;
}
- @Override
- public InternalRow apply(StructLike tuple) {
- for (int i = 0; i < types.length; i += 1) {
- Object value = tuple.get(positions[i], javaTypes[i]);
- if (value != null) {
- reusedRow.update(i, convert(value, types[i]));
- } else {
- reusedRow.setNullAt(i);
- }
- }
-
- return reusedRow;
+ public boolean isCaseSensitive() {
+ return caseSensitive;
}
- /**
- * Converts the objects into instances used by Spark's InternalRow.
- *
- * @param value a data value
- * @param type the Spark data type
- * @return the value converted to the representation expected by Spark's InternalRow.
- */
- private static Object convert(Object value, DataType type) {
- if (type instanceof StringType) {
- return UTF8String.fromString(value.toString());
- } else if (type instanceof BinaryType) {
- return ByteBuffers.toByteArray((ByteBuffer) value);
- } else if (type instanceof DecimalType) {
- return Decimal.fromDecimal(value);
+ private Schema tableSchema() {
+ if (tableSchema == null) {
+ this.tableSchema = SchemaParser.fromJson(tableSchemaString);
}
- return value;
+ return tableSchema;
}
- }
-
- public static class StructLikeInternalRow implements StructLike {
- private final DataType[] types;
- private InternalRow row = null;
- StructLikeInternalRow(StructType struct) {
- this.types = new DataType[struct.size()];
- StructField[] fields = struct.fields();
- for (int i = 0; i < fields.length; i += 1) {
- types[i] = fields[i].dataType();
+ private Schema expectedSchema() {
+ if (expectedSchema == null) {
+ this.expectedSchema = SchemaParser.fromJson(expectedSchemaString);
}
- }
-
- public StructLikeInternalRow setRow(InternalRow row) {
- this.row = row;
- return this;
- }
-
- @Override
- public int size() {
- return types.length;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public <T> T get(int pos, Class<T> javaClass) {
- return javaClass.cast(row.get(pos, types[pos]));
- }
-
- @Override
- public <T> void set(int pos, T value) {
- throw new UnsupportedOperationException("Not implemented: set");
- }
- }
-
-
- private static class IcebergRowReaderFactory implements PartitionReaderFactory {
- IcebergRowReaderFactory() {
- }
-
- @Override
- public PartitionReader<InternalRow> createReader(InputPartition inputPartition) {
- return new TaskDataReader(inputPartition);
+ return expectedSchema;
}
}
- public static class TaskDataReader implements PartitionReader<InternalRow> {
+ private static class TaskDataReader implements PartitionReader<InternalRow> {
// for some reason, the apply method can't be called from Java without reflection
private static final DynMethods.UnboundMethod APPLY_PROJECTION = DynMethods.builder("apply")
.impl(UnsafeProjection.class, InternalRow.class)
.build();
+ private final ReadTask split;
private final Iterator<FileScanTask> tasks;
- private final Schema tableSchema;
- private final Schema expectedSchema;
- private final FileIO fileIo;
private final Map<String, InputFile> inputFiles;
- private final boolean caseSensitive;
private Iterator<InternalRow> currentIterator = null;
private Closeable currentCloseable = null;
private InternalRow current = null;
- TaskDataReader(InputPartition inputPartition) {
- BatchReadInputPartition batchReadInputPartition = (BatchReadInputPartition) inputPartition;
+ TaskDataReader(ReadTask task) {
+ this.split = task;
+ this.tasks = task.files().iterator();
+ this.inputFiles = batchDecrypt(task.files());
- this.fileIo = batchReadInputPartition.fileIo;
- this.tableSchema = batchReadInputPartition.lazyTableSchema();
- this.expectedSchema = batchReadInputPartition.lazyExpectedSchema();
- this.tasks = batchReadInputPartition.task.files().iterator();
- Iterable<InputFile> decryptedFiles = batchReadInputPartition.encryptionManager.decrypt(
- Iterables.transform(batchReadInputPartition.task.files(),
- fileScanTask ->
- EncryptedFiles.encryptedInput(
- this.fileIo.newInputFile(fileScanTask.file().path().toString()),
- fileScanTask.file().keyMetadata())));
- ImmutableMap.Builder<String, InputFile> inputFileBuilder = ImmutableMap.builder();
- decryptedFiles.forEach(decrypted -> inputFileBuilder.put(decrypted.location(), decrypted));
- this.inputFiles = inputFileBuilder.build();
- // open last because the schemas and fileIo must be set
+ // open last because the split must be set
this.currentIterator = open(tasks.next());
- this.caseSensitive = batchReadInputPartition.caseSensitive;
}
@Override
@@ -526,17 +345,31 @@ public class IcebergBatchScan implements Scan,
}
}
+ private Map<String, InputFile> batchDecrypt(Collection<FileScanTask> files) {
+ Iterable<InputFile> decryptedFiles = split.encryptionManager().decrypt(
+ Iterables.transform(files,
+ fileScanTask ->
+ EncryptedFiles.encryptedInput(
+ split.io().newInputFile(fileScanTask.file().path().toString()),
+ fileScanTask.file().keyMetadata())));
+
+ ImmutableMap.Builder<String, InputFile> inputFileBuilder = ImmutableMap.builder();
+ decryptedFiles.forEach(decrypted -> inputFileBuilder.put(decrypted.location(), decrypted));
+ return inputFileBuilder.build();
+ }
+
private Iterator<InternalRow> open(FileScanTask task) {
DataFile file = task.file();
// schema or rows returned by readers
- Schema finalSchema = expectedSchema;
+ Schema finalSchema = split.expectedSchema();
PartitionSpec spec = task.spec();
Set<Integer> idColumns = spec.identitySourceIds();
// schema needed for the projection and filtering
StructType sparkType = SparkSchemaUtil.convert(finalSchema);
- Schema requiredSchema = SparkSchemaUtil.prune(tableSchema, sparkType, task.residual(), caseSensitive);
+ Schema requiredSchema = SparkSchemaUtil.prune(
+ split.tableSchema(), sparkType, task.residual(), split.isCaseSensitive());
boolean hasJoinedPartitionColumns = !idColumns.isEmpty();
boolean hasExtraFilterColumns = requiredSchema.columns().size() != finalSchema.columns().size();
@@ -647,7 +480,7 @@ public class IcebergBatchScan implements Scan,
.split(task.start(), task.length())
.createReaderFunc(fileSchema -> SparkParquetReaders.buildReader(readSchema, fileSchema))
.filter(task.residual())
- .caseSensitive(caseSensitive)
+ .caseSensitive(split.isCaseSensitive())
.build();
}
@@ -658,17 +491,114 @@ public class IcebergBatchScan implements Scan,
.schema(readSchema)
.split(task.start(), task.length())
.createReaderFunc(SparkOrcReader::new)
- .caseSensitive(caseSensitive)
+ .caseSensitive(split.isCaseSensitive())
.build();
}
private CloseableIterable<InternalRow> newDataIterable(DataTask task, Schema readSchema) {
- StructInternalRow row = new StructInternalRow(tableSchema.asStruct());
+ StructInternalRow row = new StructInternalRow(split.tableSchema().asStruct());
CloseableIterable<InternalRow> asSparkRows = CloseableIterable.transform(
task.asDataTask().rows(), row::setStruct);
return CloseableIterable.transform(
- asSparkRows, APPLY_PROJECTION.bind(projection(readSchema, tableSchema))::invoke);
+ asSparkRows, APPLY_PROJECTION.bind(projection(readSchema, split.tableSchema()))::invoke);
+ }
+ }
+
+ private static class PartitionRowConverter implements Function<StructLike, InternalRow> {
+ private final DataType[] types;
+ private final int[] positions;
+ private final Class<?>[] javaTypes;
+ private final GenericInternalRow reusedRow;
+
+ PartitionRowConverter(Schema partitionSchema, PartitionSpec spec) {
+ StructType partitionType = SparkSchemaUtil.convert(partitionSchema);
+ StructField[] fields = partitionType.fields();
+
+ this.types = new DataType[fields.length];
+ this.positions = new int[types.length];
+ this.javaTypes = new Class<?>[types.length];
+ this.reusedRow = new GenericInternalRow(types.length);
+
+ List<PartitionField> partitionFields = spec.fields();
+ for (int rowIndex = 0; rowIndex < fields.length; rowIndex += 1) {
+ this.types[rowIndex] = fields[rowIndex].dataType();
+
+ int sourceId = partitionSchema.columns().get(rowIndex).fieldId();
+ for (int specIndex = 0; specIndex < partitionFields.size(); specIndex += 1) {
+ PartitionField field = spec.fields().get(specIndex);
+ if (field.sourceId() == sourceId && "identity".equals(field.transform().toString())) {
+ positions[rowIndex] = specIndex;
+ javaTypes[rowIndex] = spec.javaClasses()[specIndex];
+ break;
+ }
+ }
+ }
+ }
+
+ @Override
+ public InternalRow apply(StructLike tuple) {
+ for (int i = 0; i < types.length; i += 1) {
+ Object value = tuple.get(positions[i], javaTypes[i]);
+ if (value != null) {
+ reusedRow.update(i, convert(value, types[i]));
+ } else {
+ reusedRow.setNullAt(i);
+ }
+ }
+
+ return reusedRow;
+ }
+
+ /**
+ * Converts the objects into instances used by Spark's InternalRow.
+ *
+ * @param value a data value
+ * @param type the Spark data type
+ * @return the value converted to the representation expected by Spark's InternalRow.
+ */
+ private static Object convert(Object value, DataType type) {
+ if (type instanceof StringType) {
+ return UTF8String.fromString(value.toString());
+ } else if (type instanceof BinaryType) {
+ return ByteBuffers.toByteArray((ByteBuffer) value);
+ } else if (type instanceof DecimalType) {
+ return Decimal.fromDecimal(value);
+ }
+ return value;
}
}
+ private static class StructLikeInternalRow implements StructLike {
+ private final DataType[] types;
+ private InternalRow row = null;
+
+ StructLikeInternalRow(StructType struct) {
+ this.types = new DataType[struct.size()];
+ StructField[] fields = struct.fields();
+ for (int i = 0; i < fields.length; i += 1) {
+ types[i] = fields[i].dataType();
+ }
+ }
+
+ public StructLikeInternalRow setRow(InternalRow row) {
+ this.row = row;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return types.length;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(row.get(pos, types[pos]));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Not implemented: set");
+ }
+ }
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchWrite.java
similarity index 90%
rename from spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchWrite.java
index dfdf6b1..b0c1ded 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchWrite.java
@@ -48,7 +48,7 @@ import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.exceptions.RuntimeIOException;
-import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
@@ -59,11 +59,11 @@ import org.apache.iceberg.spark.data.SparkParquetWriters;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.iceberg.util.Tasks;
import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.connector.catalog.TableCapability;
import org.apache.spark.sql.connector.write.BatchWrite;
import org.apache.spark.sql.connector.write.DataWriter;
import org.apache.spark.sql.connector.write.DataWriterFactory;
import org.apache.spark.sql.connector.write.WriterCommitMessage;
+import org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,31 +81,30 @@ import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
-class IcebergBatchWriter implements BatchWrite {
- private static final Logger LOG = LoggerFactory.getLogger(IcebergBatchWriter.class);
+class SparkBatchWrite implements BatchWrite {
+ private static final Logger LOG = LoggerFactory.getLogger(SparkBatchWrite.class);
private final Table table;
private final FileFormat format;
private final FileIO fileIo;
private final EncryptionManager encryptionManager;
- private final TableCapability writeBehavior;
+ private final boolean overwriteDynamic;
+ private final boolean overwriteByFilter;
+ private final Expression overwriteExpr;
private final String applicationId;
private final String wapId;
private final long targetFileSize;
private final Schema dsSchema;
- IcebergBatchWriter(
- Table table,
- CaseInsensitiveStringMap options,
- TableCapability writeBehavior,
- String applicationId,
- String wapId,
- Schema dsSchema) {
+ SparkBatchWrite(Table table, CaseInsensitiveStringMap options, boolean overwriteDynamic, boolean overwriteByFilter,
+ Expression overwriteExpr, String applicationId, String wapId, Schema dsSchema) {
this.table = table;
this.format = getFileFormat(table.properties(), options);
this.fileIo = table.io();
this.encryptionManager = table.encryption();
- this.writeBehavior = writeBehavior;
+ this.overwriteDynamic = overwriteDynamic;
+ this.overwriteByFilter = overwriteByFilter;
+ this.overwriteExpr = overwriteExpr;
this.applicationId = applicationId;
this.wapId = wapId;
this.dsSchema = dsSchema;
@@ -128,7 +127,7 @@ class IcebergBatchWriter implements BatchWrite {
}
@Override
- public DataWriterFactory createBatchWriterFactory() {
+ public WriterFactory createBatchWriterFactory() {
return new WriterFactory(
table.spec(), format, table.locationProvider(), table.properties(), fileIo, encryptionManager, targetFileSize,
dsSchema);
@@ -136,14 +135,12 @@ class IcebergBatchWriter implements BatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
- if (writeBehavior.equals(TableCapability.OVERWRITE_DYNAMIC)) {
+ if (overwriteDynamic) {
replacePartitions(messages);
- } else if (writeBehavior.equals(TableCapability.BATCH_WRITE)) {
- append(messages);
- } else if (writeBehavior.equals(TableCapability.TRUNCATE)) {
+ } else if (overwriteByFilter) {
overwrite(messages);
} else {
- throw new IllegalArgumentException("Iceberg doen't support write behavior " + writeBehavior + " for now");
+ append(messages);
}
}
@@ -192,7 +189,7 @@ class IcebergBatchWriter implements BatchWrite {
private void overwrite(WriterCommitMessage[] messages) {
OverwriteFiles overwriteFiles = table.newOverwrite();
- overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
+ overwriteFiles.overwriteByRowFilter(overwriteExpr);
int numFiles = 0;
for (DataFile file : files(messages)) {
@@ -200,7 +197,7 @@ class IcebergBatchWriter implements BatchWrite {
overwriteFiles.addFile(file);
}
- commitOperation(overwriteFiles, numFiles, "overwrite by filter or truncate");
+ commitOperation(overwriteFiles, numFiles, "overwrite by filter");
}
@Override
@@ -266,7 +263,7 @@ class IcebergBatchWriter implements BatchWrite {
}
}
- protected static class WriterFactory implements DataWriterFactory {
+ private static class WriterFactory implements DataWriterFactory, StreamingDataWriterFactory {
private final PartitionSpec spec;
private final FileFormat format;
private final LocationProvider locations;
@@ -276,15 +273,9 @@ class IcebergBatchWriter implements BatchWrite {
private final long targetFileSize;
private final Schema dsSchema;
- WriterFactory(
- PartitionSpec spec,
- FileFormat format,
- LocationProvider locations,
- Map<String, String> properties,
- FileIO fileIo,
- EncryptionManager encryptionManager,
- long targetFileSize,
- Schema dsSchema) {
+ WriterFactory(PartitionSpec spec, FileFormat format, LocationProvider locations,
+ Map<String, String> properties, FileIO fileIo, EncryptionManager encryptionManager,
+ long targetFileSize, Schema dsSchema) {
this.spec = spec;
this.format = format;
this.locations = locations;
@@ -295,20 +286,13 @@ class IcebergBatchWriter implements BatchWrite {
this.dsSchema = dsSchema;
}
- public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
- OutputFileFactory fileFactory = new OutputFileFactory(partitionId, taskId, epochId);
- AppenderFactory<InternalRow> appenderFactory = new SparkAppenderFactory();
- if (spec.fields().isEmpty()) {
- return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
- } else {
- return new PartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
- }
+ public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
+ return createWriter(partitionId, taskId, 0);
}
- public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
- OutputFileFactory fileFactory = new OutputFileFactory(partitionId, taskId, 0);
+ public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
+ OutputFileFactory fileFactory = new OutputFileFactory(partitionId, taskId, epochId);
AppenderFactory<InternalRow> appenderFactory = new SparkAppenderFactory();
-
if (spec.fields().isEmpty()) {
return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
} else {
@@ -406,13 +390,8 @@ class IcebergBatchWriter implements BatchWrite {
private EncryptedOutputFile currentFile = null;
private long currentRows = 0;
- BaseWriter(
- PartitionSpec spec,
- FileFormat format,
- AppenderFactory<InternalRow> appenderFactory,
- WriterFactory.OutputFileFactory fileFactory,
- FileIO fileIo,
- long targetFileSize) {
+ BaseWriter(PartitionSpec spec, FileFormat format, AppenderFactory<InternalRow> appenderFactory,
+ WriterFactory.OutputFileFactory fileFactory, FileIO fileIo, long targetFileSize) {
this.spec = spec;
this.format = format;
this.appenderFactory = appenderFactory;
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
new file mode 100644
index 0000000..cb932ff
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -0,0 +1,102 @@
+/*
+ * 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.iceberg.spark.source;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.spark.SparkFilters;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
+import org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns {
+ private static final Filter[] NO_FILTERS = new Filter[0];
+
+ private final SparkSession spark;
+ private final Table table;
+ private final CaseInsensitiveStringMap options;
+
+ private Schema schema;
+ private boolean caseSensitive;
+ private List<Expression> filterExpressions = null;
+ private Filter[] pushedFilters = NO_FILTERS;
+
+ SparkScanBuilder(SparkSession spark, Table table, CaseInsensitiveStringMap options) {
+ this.spark = spark;
+ this.table = table;
+ this.options = options;
+ this.schema = table.schema();
+ this.caseSensitive = Boolean.parseBoolean(spark.conf().get("spark.sql.caseSensitive"));
+ }
+
+ public SparkScanBuilder caseSensitive(boolean isCaseSensitive) {
+ this.caseSensitive = isCaseSensitive;
+ return this;
+ }
+
+ @Override
+ public Filter[] pushFilters(Filter[] filters) {
+ List<Expression> expressions = Lists.newArrayListWithExpectedSize(filters.length);
+ List<Filter> pushed = Lists.newArrayListWithExpectedSize(filters.length);
+
+ for (Filter filter : filters) {
+ Expression expr = SparkFilters.convert(filter);
+ if (expr != null) {
+ expressions.add(expr);
+ pushed.add(filter);
+ }
+ }
+
+ this.filterExpressions = expressions;
+ this.pushedFilters = pushed.toArray(new Filter[0]);
+
+ // Spark doesn't support residuals per task, so return all filters
+ // to get Spark to handle record-level filtering
+ return filters;
+ }
+
+ @Override
+ public Filter[] pushedFilters() {
+ return pushedFilters;
+ }
+
+ @Override
+ public void pruneColumns(StructType requestedSchema) {
+ this.schema = SparkSchemaUtil.prune(table.schema(), requestedSchema);
+ }
+
+ public StructType readSchema() {
+ return SparkSchemaUtil.convert(schema);
+ }
+
+ @Override
+ public Scan build() {
+ return new SparkBatchScan(table, caseSensitive, schema, filterExpressions, options);
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkStreamingWrite.java
similarity index 64%
rename from spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/SparkStreamingWrite.java
index e1d0dfb..5781534 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkStreamingWrite.java
@@ -35,7 +35,6 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.connector.catalog.TableCapability;
import org.apache.spark.sql.connector.write.DataWriter;
import org.apache.spark.sql.connector.write.WriterCommitMessage;
import org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory;
@@ -44,58 +43,31 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
-import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
+public class SparkStreamingWrite extends SparkBatchWrite implements StreamingWrite {
-public class IcebergStreamingWriter extends IcebergBatchWriter implements StreamingWrite {
-
- private static final Logger LOG = LoggerFactory.getLogger(IcebergStreamingWriter.class);
+ private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingWrite.class);
private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";
private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";
+ private final boolean truncateBatches;
private final String queryId;
- private final TableCapability writeBehavior;
- private final Table table;
- private final long targetFileSize;
- private final FileFormat format;
- private final Schema dsSchema;
- IcebergStreamingWriter(Table table, CaseInsensitiveStringMap options, String queryId, TableCapability writeBehavior,
- String applicationId, String wapId, Schema dsSchema) {
- super(table, options, writeBehavior, applicationId, wapId, dsSchema);
+ SparkStreamingWrite(Table table, CaseInsensitiveStringMap options, boolean truncateBatches, String queryId,
+ String applicationId, String wapId, Schema dsSchema) {
+ super(table, options, false, truncateBatches, Expressions.alwaysTrue(), applicationId, wapId, dsSchema);
+ this.truncateBatches = truncateBatches;
this.queryId = queryId;
- this.writeBehavior = writeBehavior;
- this.table = table;
- this.format = getFileFormat(table.properties(), options);
- long tableTargetFileSize = PropertyUtil.propertyAsLong(
- table.properties(), WRITE_TARGET_FILE_SIZE_BYTES, WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
- this.targetFileSize = options.getLong("target-file-size-bytes", tableTargetFileSize);
- this.dsSchema = dsSchema;
}
@Override
public StreamingDataWriterFactory createStreamingWriterFactory() {
- return new StreamingWriterFactory(table.spec(), format, table.locationProvider(),
- table.properties(), table.io(), table.encryption(), targetFileSize, dsSchema);
- }
-
- private static class StreamingWriterFactory extends WriterFactory implements StreamingDataWriterFactory {
-
- StreamingWriterFactory(PartitionSpec spec, FileFormat format, LocationProvider locations,
- Map<String, String> properties, FileIO fileIo, EncryptionManager encryptionManager,
- long targetFileSize, Schema dsSchema) {
- super(spec, format, locations, properties, fileIo, encryptionManager, targetFileSize, dsSchema);
- }
-
- @Override
- public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
- return super.createWriter(partitionId, taskId, epochId);
- }
+ // the writer factory works for both batch and streaming
+ return createBatchWriterFactory();
}
@Override
public void commit(long epochId, WriterCommitMessage[] messages) {
- LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, writeBehavior);
+ LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, truncateBatches ? "complete" : "append");
table().refresh();
Long lastCommittedEpochId = getLastCommittedEpochId();
@@ -104,7 +76,7 @@ public class IcebergStreamingWriter extends IcebergBatchWriter implements Stream
return;
}
- if (writeBehavior.equals(TableCapability.TRUNCATE)) {
+ if (truncateBatches) {
OverwriteFiles overwriteFiles = table().newOverwrite();
overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
int numFiles = 0;
@@ -113,7 +85,7 @@ public class IcebergStreamingWriter extends IcebergBatchWriter implements Stream
numFiles++;
}
commit(overwriteFiles, epochId, numFiles, "streaming complete overwrite");
- } else if (writeBehavior.equals(TableCapability.STREAMING_WRITE)) {
+ } else {
AppendFiles append = table().newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
@@ -121,8 +93,6 @@ public class IcebergStreamingWriter extends IcebergBatchWriter implements Stream
numFiles++;
}
commit(append, epochId, numFiles, "streaming append");
- } else {
- throw new IllegalArgumentException("Iceberg doen't support write behavior " + writeBehavior + " for now");
}
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
new file mode 100644
index 0000000..f85aac2
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -0,0 +1,118 @@
+/*
+ * 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.iceberg.spark.source;
+
+import com.google.common.collect.ImmutableSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.spark.SparkUtil;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+import org.apache.spark.sql.connector.catalog.SupportsWrite;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+class SparkTable implements org.apache.spark.sql.connector.catalog.Table, SupportsRead, SupportsWrite {
+
+ private static final Set<TableCapability> CAPABILITIES = ImmutableSet.of(
+ TableCapability.BATCH_READ,
+ TableCapability.BATCH_WRITE,
+ TableCapability.STREAMING_WRITE,
+ TableCapability.OVERWRITE_BY_FILTER,
+ TableCapability.OVERWRITE_DYNAMIC);
+
+ private final Table icebergTable;
+ private final StructType requestedSchema;
+ private StructType lazyTableSchema = null;
+ private SparkSession lazySpark = null;
+
+ SparkTable(Table icebergTable, StructType requestedSchema) {
+ this.icebergTable = icebergTable;
+ this.requestedSchema = requestedSchema;
+
+ if (requestedSchema != null) {
+ // convert the requested schema to throw an exception if any requested fields are unknown
+ SparkSchemaUtil.convert(icebergTable.schema(), requestedSchema);
+ }
+ }
+
+ private SparkSession sparkSession() {
+ if (lazySpark == null) {
+ this.lazySpark = SparkSession.active();
+ }
+
+ return lazySpark;
+ }
+
+ @Override
+ public String name() {
+ return icebergTable.toString();
+ }
+
+ @Override
+ public StructType schema() {
+ if (lazyTableSchema == null) {
+ if (requestedSchema != null) {
+ this.lazyTableSchema = SparkSchemaUtil.convert(SparkSchemaUtil.prune(icebergTable.schema(), requestedSchema));
+ } else {
+ this.lazyTableSchema = SparkSchemaUtil.convert(icebergTable.schema());
+ }
+ }
+
+ return lazyTableSchema;
+ }
+
+ @Override
+ public Transform[] partitioning() {
+ return SparkUtil.toTransforms(icebergTable.spec());
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return icebergTable.properties();
+ }
+
+ @Override
+ public Set<TableCapability> capabilities() {
+ return CAPABILITIES;
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
+ SparkScanBuilder scanBuilder = new SparkScanBuilder(sparkSession(), icebergTable, options);
+ if (requestedSchema != null) {
+ scanBuilder.pruneColumns(requestedSchema);
+ }
+
+ return scanBuilder;
+ }
+
+ @Override
+ public WriteBuilder newWriteBuilder(CaseInsensitiveStringMap options) {
+ return new SparkWriteBuilder(sparkSession(), icebergTable, options);
+ }
+
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java
new file mode 100644
index 0000000..3f1b48b
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkWriteBuilder.java
@@ -0,0 +1,173 @@
+/*
+ * 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.iceberg.spark.source;
+
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.Locale;
+import java.util.stream.Collectors;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.spark.SparkFilters;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.transforms.UnknownTransform;
+import org.apache.iceberg.types.CheckCompatibility;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.write.BatchWrite;
+import org.apache.spark.sql.connector.write.SupportsDynamicOverwrite;
+import org.apache.spark.sql.connector.write.SupportsOverwrite;
+import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.connector.write.streaming.StreamingWrite;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+class SparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
+
+ private final SparkSession spark;
+ private final Table table;
+ private final CaseInsensitiveStringMap options;
+ private final String overwriteMode;
+ private boolean overwriteDynamic = false;
+ private boolean overwriteByFilter = false;
+ private Expression overwriteExpr = null;
+ private String writeQueryId = null;
+ private StructType writeSchema = null;
+
+ SparkWriteBuilder(SparkSession spark, Table table, CaseInsensitiveStringMap options) {
+ this.spark = spark;
+ this.table = table;
+ this.options = options;
+ this.overwriteMode = options.containsKey("overwrite-mode") ?
+ options.get("overwrite-mode").toLowerCase(Locale.ROOT) : null;
+ }
+
+ @Override
+ public WriteBuilder withQueryId(String queryId) {
+ this.writeQueryId = queryId;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder withInputDataSchema(StructType schemaInput) {
+ this.writeSchema = schemaInput;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder overwriteDynamicPartitions() {
+ Preconditions.checkState(!overwriteByFilter, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
+ this.overwriteDynamic = true;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder overwrite(Filter[] filters) {
+ this.overwriteExpr = SparkFilters.convert(filters);
+ if (overwriteExpr == Expressions.alwaysTrue() && "dynamic".equals(overwriteMode)) {
+ // use the write option to override truncating the table. use dynamic overwrite instead.
+ this.overwriteDynamic = true;
+ } else {
+ Preconditions.checkState(!overwriteDynamic, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
+ this.overwriteByFilter = true;
+ }
+ return this;
+ }
+
+ @Override
+ public BatchWrite buildForBatch() {
+ // Validate
+ Schema dsSchema = SparkSchemaUtil.convert(table.schema(), writeSchema);
+ validateWriteSchema(table.schema(), dsSchema, checkNullability(options));
+ validatePartitionTransforms(table.spec());
+
+ // Get application id
+ String appId = spark.sparkContext().applicationId();
+
+ // Get write-audit-publish id
+ String wapId = spark.conf().get("spark.wap.id", null);
+
+ return new SparkBatchWrite(table,
+ options, overwriteDynamic, overwriteByFilter, overwriteExpr, appId, wapId, dsSchema);
+ }
+
+ @Override
+ public StreamingWrite buildForStreaming() {
+ // Validate
+ Schema dsSchema = SparkSchemaUtil.convert(table.schema(), writeSchema);
+ validateWriteSchema(table.schema(), dsSchema, checkNullability(options));
+ validatePartitionTransforms(table.spec());
+
+ // Change to streaming write if it is just append
+ Preconditions.checkState(!overwriteDynamic,
+ "Unsupported streaming operation: dynamic partition overwrite");
+ Preconditions.checkState(!overwriteByFilter || overwriteExpr == Expressions.alwaysTrue(),
+ "Unsupported streaming operation: overwrite by filter: %s", overwriteExpr);
+
+ // Get application id
+ String appId = spark.sparkContext().applicationId();
+ String wapId = spark.conf().get("spark.wap.id", null);
+
+ return new SparkStreamingWrite(table, options, overwriteByFilter, writeQueryId, appId, wapId, dsSchema);
+ }
+
+ private void validateWriteSchema(Schema tableSchema, Schema dsSchema, Boolean checkNullability) {
+ List<String> errors;
+ if (checkNullability) {
+ errors = CheckCompatibility.writeCompatibilityErrors(tableSchema, dsSchema);
+ } else {
+ errors = CheckCompatibility.typeCompatibilityErrors(tableSchema, dsSchema);
+ }
+ if (!errors.isEmpty()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Cannot write incompatible dataset to table with schema:\n")
+ .append(tableSchema)
+ .append("\nProblems:");
+ for (String error : errors) {
+ sb.append("\n* ").append(error);
+ }
+ throw new IllegalArgumentException(sb.toString());
+ }
+ }
+
+ private void validatePartitionTransforms(PartitionSpec spec) {
+ if (spec.fields().stream().anyMatch(field -> field.transform() instanceof UnknownTransform)) {
+ String unsupported = spec.fields().stream()
+ .map(PartitionField::transform)
+ .filter(transform -> transform instanceof UnknownTransform)
+ .map(org.apache.iceberg.transforms.Transform::toString)
+ .collect(Collectors.joining(", "));
+
+ throw new UnsupportedOperationException(
+ String.format("Cannot write using unsupported transforms: %s", unsupported));
+ }
+ }
+
+ private boolean checkNullability(CaseInsensitiveStringMap options) {
+ boolean sparkCheckNullability = Boolean.parseBoolean(spark.conf()
+ .get("spark.sql.iceberg.check-nullability", "true"));
+ boolean dataFrameCheckNullability = options.getBoolean("check-nullability", true);
+ return sparkCheckNullability && dataFrameCheckNullability;
+ }
+}
diff --git a/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
index 04e5f32..01a6c4e 100644
--- a/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
+++ b/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -17,4 +17,4 @@
# under the License.
#
-org.apache.iceberg.spark.source.IcebergTableProvider
+org.apache.iceberg.spark.source.IcebergSource
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
index 9ff583d..85d2627 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
@@ -53,8 +53,8 @@ import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.InternalRow;
import org.junit.AfterClass;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -221,9 +221,10 @@ public class TestDataFrameWrites extends AvroDataTest {
return spark.internalCreateDataFrame(JavaRDD.toRDD(rdd), convert(schema), false);
}
- // This fails due to SPARK-28730
- @Ignore
+ @Test
public void testNullableWithWriteOption() throws IOException {
+ Assume.assumeTrue("Spark 3.0 rejects writing nulls to a required column", spark.version().startsWith("2"));
+
File location = new File(temp.newFolder("parquet"), "test");
String sourcePath = String.format("%s/nullable_poc/sourceFolder/", location.toString());
String targetPath = String.format("%s/nullable_poc/targetFolder/", location.toString());
@@ -266,9 +267,10 @@ public class TestDataFrameWrites extends AvroDataTest {
}
- // This fails due to SPARK-28730
- @Ignore
+ @Test
public void testNullableWithSparkSqlOption() throws IOException {
+ Assume.assumeTrue("Spark 3.0 rejects writing nulls to a required column", spark.version().startsWith("2"));
+
File location = new File(temp.newFolder("parquet"), "test");
String sourcePath = String.format("%s/nullable_poc/sourceFolder/", location.toString());
String targetPath = String.format("%s/nullable_poc/targetFolder/", location.toString());
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 991e01d..ead9a27 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -51,8 +51,10 @@ import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.connector.read.Batch;
import org.apache.spark.sql.connector.read.InputPartition;
import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.ScanBuilder;
import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
import org.apache.spark.sql.sources.And;
import org.apache.spark.sql.sources.EqualTo;
@@ -212,10 +214,11 @@ public class TestFilteredScan {
CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of(
"path", unpartitioned.toString())
);
- IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
for (int i = 0; i < 10; i += 1) {
- pushFilters(scan, EqualTo.apply("id", i));
+ pushFilters(builder, EqualTo.apply("id", i));
+ Batch scan = builder.build().toBatch();
InputPartition[] partitions = scan.planInputPartitions();
Assert.assertEquals("Should only create one task for a small file", 1, partitions.length);
@@ -239,9 +242,11 @@ public class TestFilteredScan {
try {
for (int i = 0; i < 10; i += 1) {
- IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), false, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options)
+ .caseSensitive(false);
- pushFilters(scan, EqualTo.apply("ID", i)); // note lower(ID) == lower(id), so there must be a match
+ pushFilters(builder, EqualTo.apply("ID", i)); // note lower(ID) == lower(id), so there must be a match
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should only create one task for a small file", 1, tasks.length);
@@ -262,9 +267,10 @@ public class TestFilteredScan {
"path", unpartitioned.toString())
);
- IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(builder, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should only create one task for a small file", 1, tasks.length);
@@ -278,14 +284,15 @@ public class TestFilteredScan {
Table table = buildPartitionedTable("bucketed_by_id", BUCKET_BY_ID, "bucket4", "id");
CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
+ Batch unfiltered = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options).build().toBatch();
Assert.assertEquals("Unfiltered table should created 4 read tasks",
4, unfiltered.planInputPartitions().length);
for (int i = 0; i < 10; i += 1) {
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, EqualTo.apply("id", i));
+ pushFilters(builder, EqualTo.apply("id", i));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
@@ -302,15 +309,16 @@ public class TestFilteredScan {
public void testDayPartitionedTimestampFilters() {
Table table = buildPartitionedTable("partitioned_by_day", PARTITION_BY_DAY, "ts_day", "ts");
CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
+ Batch unfiltered = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options).build().toBatch();
Assert.assertEquals("Unfiltered table should created 2 read tasks",
2, unfiltered.planInputPartitions().length);
{
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(builder, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should create one task for 2017-12-21", 1, tasks.length);
@@ -320,11 +328,12 @@ public class TestFilteredScan {
}
{
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, And.apply(
+ pushFilters(builder, And.apply(
GreaterThan.apply("ts", "2017-12-22T06:00:00+00:00"),
LessThan.apply("ts", "2017-12-22T08:00:00+00:00")));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should create one task for 2017-12-22", 1, tasks.length);
@@ -341,29 +350,31 @@ public class TestFilteredScan {
Table table = buildPartitionedTable("partitioned_by_hour", PARTITION_BY_HOUR, "ts_hour", "ts");
CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
+ Batch unfiltered = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options).build().toBatch();
Assert.assertEquals("Unfiltered table should created 9 read tasks",
9, unfiltered.planInputPartitions().length);
{
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(builder, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should create 4 tasks for 2017-12-21: 15, 17, 21, 22", 4, tasks.length);
assertEqualsSafe(SCHEMA.asStruct(), expected(8, 9, 7, 6, 5),
- read(table.location().toString(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
+ read(table.location(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
}
{
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, And.apply(
+ pushFilters(builder, And.apply(
GreaterThan.apply("ts", "2017-12-22T06:00:00+00:00"),
LessThan.apply("ts", "2017-12-22T08:00:00+00:00")));
+ Batch scan = builder.build().toBatch();
InputPartition[] tasks = scan.planInputPartitions();
Assert.assertEquals("Should create 2 tasks for 2017-12-22: 6, 7", 2, tasks.length);
@@ -412,9 +423,10 @@ public class TestFilteredScan {
Table table = buildPartitionedTable("partitioned_by_data", PARTITION_BY_DATA, "data_ident", "data");
CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
- pushFilters(scan, new StringStartsWith("data", "junc"));
+ pushFilters(builder, new StringStartsWith("data", "junc"));
+ Batch scan = builder.build().toBatch();
Assert.assertEquals(1, scan.planInputPartitions().length);
}
@@ -427,8 +439,10 @@ public class TestFilteredScan {
"path", table.location())
);
- IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(scan, new StringStartsWith("data", "junc"));
+ SparkScanBuilder builder = new SparkScanBuilder(spark, TABLES.load(options.get("path")), options);
+
+ pushFilters(builder, new StringStartsWith("data", "junc"));
+ Batch scan = builder.build().toBatch();
Assert.assertEquals(1, scan.planInputPartitions().length);
}
@@ -487,7 +501,7 @@ public class TestFilteredScan {
return expected;
}
- private void pushFilters(Scan scan, Filter... filters) {
+ private void pushFilters(ScanBuilder scan, Filter... filters) {
Assert.assertTrue(scan instanceof SupportsPushDownFilters);
SupportsPushDownFilters filterable = (SupportsPushDownFilters) scan;
filterable.pushFilters(filters);
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
index fdf5daa..dd820fa 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
@@ -20,16 +20,17 @@
package org.apache.iceberg.spark.source;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Table;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-public class TestIcebergSource extends IcebergTableProvider {
+public class TestIcebergSource extends IcebergSource {
@Override
public String shortName() {
return "iceberg-test";
}
@Override
- public org.apache.iceberg.Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
+ protected Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
return TestTables.load(options.get("iceberg.table.name"));
}
}
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
index da936ed..31ce025 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
@@ -39,8 +39,8 @@ import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.junit.AfterClass;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -164,8 +164,7 @@ public class TestParquetWrite {
Assert.assertEquals("Result rows should match", expected, actual);
}
- // ignore due to spark default use static
- @Ignore
+ @Test
public void testOverwrite() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
@@ -197,7 +196,7 @@ public class TestParquetWrite {
// overwrite with 2*id to replace record 2, append 4 and 6
df.withColumn("id", df.col("id").multiply(2)).select("id", "data").write()
- .option("partitionOverwriteMode", "dynamic")
+ .option("overwrite-mode", "dynamic")
.format("iceberg")
.mode("overwrite")
.save(location.toString());
@@ -342,9 +341,12 @@ public class TestParquetWrite {
Assert.assertTrue("All DataFiles contain 1000 rows", files.stream().allMatch(d -> d.recordCount() == 1000));
}
- // This fails due to SPARK-28730
- @Ignore
+ @Test
public void testWriteProjection() throws IOException {
+ Assume.assumeTrue(
+ "Not supported in Spark 3.0; analysis requires all columns are present",
+ spark.version().startsWith("2"));
+
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
@@ -376,9 +378,12 @@ public class TestParquetWrite {
Assert.assertEquals("Result rows should match", expected, actual);
}
- // This fails due to SPARK-28730
- @Ignore
+ @Test
public void testWriteProjectionWithMiddle() throws IOException {
+ Assume.assumeTrue(
+ "Not supported in Spark 3.0; analysis requires all columns are present",
+ spark.version().startsWith("2"));
+
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
index 9550d00..b32ce7f 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
@@ -41,7 +41,6 @@ import org.apache.spark.sql.streaming.StreamingQueryException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -200,8 +199,7 @@ public class TestStructuredStreaming {
}
}
- // This fails due to SPARK-28730
- @Ignore
+ @Test
public void testStreamingWriteCompleteModeWithProjection() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test-table");
[incubator-iceberg] 01/03: Supports spark-3.0 data source V2 APIs
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch spark-3
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
commit 1112066a7ef3e5a858d6c6705efa45cab001d2a8
Author: jimmyjchen <ji...@tencent.com>
AuthorDate: Tue Nov 26 21:55:26 2019 +0800
Supports spark-3.0 data source V2 APIs
---
api/src/main/java/org/apache/iceberg/Table.java | 9 +
build.gradle | 10 +-
.../java/org/apache/iceberg/spark/SparkUtils.java | 92 ++++
.../source/{Reader.java => IcebergBatchScan.java} | 469 +++++++++++----------
.../{Writer.java => IcebergBatchWriter.java} | 98 +++--
.../apache/iceberg/spark/source/IcebergSource.java | 204 ---------
...mingWriter.java => IcebergStreamingWriter.java} | 72 +++-
.../apache/iceberg/spark/source/IcebergTable.java | 222 ++++++++++
.../iceberg/spark/source/IcebergTableProvider.java | 101 +++++
.../org/apache/iceberg/spark/source/Stats.java | 2 +-
...org.apache.spark.sql.sources.DataSourceRegister | 2 +-
.../iceberg/spark/data/TestSparkDateTimes.java | 7 +-
.../iceberg/spark/source/TestDataFrameWrites.java | 7 +-
.../iceberg/spark/source/TestFilteredScan.java | 174 ++++----
.../iceberg/spark/source/TestIcebergSource.java | 9 +-
.../iceberg/spark/source/TestParquetWrite.java | 11 +-
.../spark/source/TestStructuredStreaming.java | 9 +-
versions.lock | 270 ++++++------
versions.props | 4 +-
19 files changed, 1061 insertions(+), 711 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/Table.java b/api/src/main/java/org/apache/iceberg/Table.java
index 63c55e5..e0fa525 100644
--- a/api/src/main/java/org/apache/iceberg/Table.java
+++ b/api/src/main/java/org/apache/iceberg/Table.java
@@ -228,4 +228,13 @@ public interface Table {
* @return a {@link LocationProvider} to provide locations for new data files
*/
LocationProvider locationProvider();
+
+ /**
+ * Return the name of this table.
+ *
+ * @return this table's name
+ */
+ default String name() {
+ return "table(" + hashCode() + ")";
+ }
}
diff --git a/build.gradle b/build.gradle
index a1f176c..6835cbe 100644
--- a/build.gradle
+++ b/build.gradle
@@ -240,7 +240,13 @@ project(':iceberg-spark') {
compile project(':iceberg-hive')
compileOnly "org.apache.avro:avro"
- compileOnly("org.apache.spark:spark-hive_2.11") {
+ compileOnly("org.apache.spark:spark-sql_2.12") {
+ exclude group: 'org.apache.avro', module: 'avro'
+ }
+ compileOnly("org.apache.spark:spark-hive_2.12") {
+ exclude group: 'org.apache.avro', module: 'avro'
+ }
+ compileOnly("org.apache.spark:spark-catalyst_2.12") {
exclude group: 'org.apache.avro', module: 'avro'
}
@@ -256,7 +262,7 @@ project(':iceberg-spark') {
// dependency only to the jmh configuration, however gradle-consistent-versions
// plugin does not respect this configuration and does not seem to have a way
// to add custom configurations in its lockable configuration detection
- compileOnly("org.apache.spark:spark-avro_2.11") {
+ compileOnly("org.apache.spark:spark-avro_2.12") {
exclude group: 'org.apache.avro', module: 'avro'
}
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java b/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java
new file mode 100644
index 0000000..64e81cb
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkUtils.java
@@ -0,0 +1,92 @@
+/*
+ * 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.iceberg.spark;
+
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.expressions.Expressions;
+import org.apache.spark.sql.connector.expressions.Transform;
+
+public final class SparkUtils {
+
+ private static final Pattern HAS_WIDTH = Pattern.compile("(\\w+)\\[(\\d+)\\]");
+
+ private SparkUtils() {}
+
+ public static SparkSession getSparkSession() {
+ return SparkSession.builder().getOrCreate();
+ }
+
+ public static Configuration getBaseConf() {
+ return getSparkSession().sparkContext().hadoopConfiguration();
+ }
+
+
+ public static Transform[] toTransforms(PartitionSpec spec) {
+ List<Transform> transforms = Lists.newArrayList();
+ int numBuckets = 0;
+ List<String> bucketColumns = Lists.newArrayList();
+
+ for (PartitionField f : spec.fields()) {
+ Matcher widthMatcher = HAS_WIDTH.matcher(f.transform().toString());
+ if (widthMatcher.matches()) {
+ String name = widthMatcher.group(1);
+ if (name.equalsIgnoreCase("truncate")) {
+ throw new UnsupportedOperationException("Spark doesn't support truncate transform");
+
+ } else if (name.equalsIgnoreCase("bucket")) {
+ numBuckets = Integer.parseInt(widthMatcher.group(2));
+ bucketColumns.add(spec.schema().findColumnName(f.sourceId()));
+
+ } else if (f.transform().toString().equalsIgnoreCase("identity")) {
+ transforms.add(Expressions.identity(spec.schema().findColumnName(f.sourceId())));
+
+ } else if (f.transform().toString().equalsIgnoreCase("year")) {
+ transforms.add(Expressions.years(spec.schema().findColumnName(f.sourceId())));
+
+ } else if (f.transform().toString().equalsIgnoreCase("month")) {
+ transforms.add(Expressions.months(spec.schema().findColumnName(f.sourceId())));
+
+ } else if (f.transform().toString().equalsIgnoreCase("day")) {
+ transforms.add(Expressions.days(spec.schema().findColumnName(f.sourceId())));
+
+ } else if (f.transform().toString().equalsIgnoreCase("hour")) {
+ transforms.add(Expressions.hours(spec.schema().findColumnName(f.sourceId())));
+
+ } else {
+ throw new UnsupportedOperationException("Spark doesn't support transform " + f.transform());
+ }
+ }
+ }
+
+ if (!bucketColumns.isEmpty()) {
+ transforms.add(Expressions.bucket(numBuckets, bucketColumns.toArray(new String[0])));
+ }
+
+ return transforms.toArray(new Transform[0]);
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java
similarity index 82%
rename from spark/src/main/java/org/apache/iceberg/spark/source/Reader.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java
index 43c966a..b3fef6d 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchScan.java
@@ -70,15 +70,16 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference;
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
import org.apache.spark.sql.catalyst.expressions.JoinedRow;
import org.apache.spark.sql.catalyst.expressions.UnsafeProjection;
+import org.apache.spark.sql.connector.read.Batch;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.Statistics;
+import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
+import org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns;
+import org.apache.spark.sql.connector.read.SupportsReportStatistics;
import org.apache.spark.sql.sources.Filter;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
-import org.apache.spark.sql.sources.v2.reader.InputPartition;
-import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
-import org.apache.spark.sql.sources.v2.reader.Statistics;
-import org.apache.spark.sql.sources.v2.reader.SupportsPushDownFilters;
-import org.apache.spark.sql.sources.v2.reader.SupportsPushDownRequiredColumns;
-import org.apache.spark.sql.sources.v2.reader.SupportsReportStatistics;
import org.apache.spark.sql.types.BinaryType;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.Decimal;
@@ -86,15 +87,15 @@ import org.apache.spark.sql.types.DecimalType;
import org.apache.spark.sql.types.StringType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.apache.spark.unsafe.types.UTF8String;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import scala.collection.JavaConverters;
-class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushDownRequiredColumns,
+public class IcebergBatchScan implements Scan,
+ Batch,
+ SupportsPushDownFilters,
+ SupportsPushDownRequiredColumns,
SupportsReportStatistics {
- private static final Logger LOG = LoggerFactory.getLogger(Reader.class);
-
private static final Filter[] NO_FILTERS = new Filter[0];
private final Table table;
@@ -115,19 +116,23 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private StructType type = null; // cached because Spark accesses it multiple times
private List<CombinedScanTask> tasks = null; // lazy cache of tasks
- Reader(Table table, boolean caseSensitive, DataSourceOptions options) {
+ public IcebergBatchScan(Table table, Boolean caseSensitive, CaseInsensitiveStringMap options) {
this.table = table;
- this.snapshotId = options.get("snapshot-id").map(Long::parseLong).orElse(null);
- this.asOfTimestamp = options.get("as-of-timestamp").map(Long::parseLong).orElse(null);
+ this.snapshotId = options.containsKey("snapshot-id") ? options.getLong("snapshot-id", 0) : null;
+ this.asOfTimestamp = options.containsKey("as-of-timestamp") ? options.getLong("as-of-timestamp", 0) : null;
+
if (snapshotId != null && asOfTimestamp != null) {
throw new IllegalArgumentException(
"Cannot scan using both snapshot-id and as-of-timestamp to select the table snapshot");
}
// look for split behavior overrides in options
- this.splitSize = options.get("split-size").map(Long::parseLong).orElse(null);
- this.splitLookback = options.get("lookback").map(Integer::parseInt).orElse(null);
- this.splitOpenFileCost = options.get("file-open-cost").map(Long::parseLong).orElse(null);
+ this.splitSize = options.containsKey("split-size") ? options.getLong("split-size",
+ TableProperties.SPLIT_SIZE_DEFAULT) : null;
+ this.splitLookback = options.containsKey("lookback") ? options.getInt("lookback",
+ TableProperties.SPLIT_LOOKBACK_DEFAULT) : null;
+ this.splitOpenFileCost = options.containsKey("file-open-cost") ? options.getLong("file-open-cost",
+ TableProperties.SPLIT_OPEN_FILE_COST_DEFAULT) : null;
this.schema = table.schema();
this.fileIo = table.io();
@@ -135,41 +140,59 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
this.caseSensitive = caseSensitive;
}
- private Schema lazySchema() {
- if (schema == null) {
- if (requestedSchema != null) {
- this.schema = SparkSchemaUtil.prune(table.schema(), requestedSchema);
- } else {
- this.schema = table.schema();
- }
- }
- return schema;
+ public IcebergBatchScan(Table table, CaseInsensitiveStringMap options) {
+ this(table, true, options);
}
- private StructType lazyType() {
- if (type == null) {
- this.type = SparkSchemaUtil.convert(lazySchema());
+ public IcebergBatchScan(Table table, CaseInsensitiveStringMap options, StructType requestedSchema) {
+ this(table, true, options);
+
+ if (requestedSchema != null) {
+ pruneColumns(requestedSchema);
}
- return type;
}
- @Override
- public StructType readSchema() {
- return lazyType();
- }
+ private List<CombinedScanTask> tasks() {
+ if (tasks == null) {
+ TableScan scan = table
+ .newScan()
+ .caseSensitive(caseSensitive)
+ .project(lazySchema());
- @Override
- public List<InputPartition<InternalRow>> planInputPartitions() {
- String tableSchemaString = SchemaParser.toJson(table.schema());
- String expectedSchemaString = SchemaParser.toJson(lazySchema());
+ if (snapshotId != null) {
+ scan = scan.useSnapshot(snapshotId);
+ }
- List<InputPartition<InternalRow>> readTasks = Lists.newArrayList();
- for (CombinedScanTask task : tasks()) {
- readTasks.add(
- new ReadTask(task, tableSchemaString, expectedSchemaString, fileIo, encryptionManager, caseSensitive));
+ if (asOfTimestamp != null) {
+ scan = scan.asOfTime(asOfTimestamp);
+ }
+
+ if (splitSize != null) {
+ scan = scan.option(TableProperties.SPLIT_SIZE, splitSize.toString());
+ }
+
+ if (splitLookback != null) {
+ scan = scan.option(TableProperties.SPLIT_LOOKBACK, splitLookback.toString());
+ }
+
+ if (splitOpenFileCost != null) {
+ scan = scan.option(TableProperties.SPLIT_OPEN_FILE_COST, splitOpenFileCost.toString());
+ }
+
+ if (filterExpressions != null) {
+ for (Expression filter : filterExpressions) {
+ scan = scan.filter(filter);
+ }
+ }
+
+ try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
+ this.tasks = Lists.newArrayList(tasksIterable);
+ } catch (IOException e) {
+ throw new RuntimeIOException(e, "Failed to close table scan: %s", scan);
+ }
}
- return readTasks;
+ return tasks;
}
@Override
@@ -200,6 +223,11 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
}
@Override
+ public Batch toBatch() {
+ return this;
+ }
+
+ @Override
public Filter[] pushedFilters() {
return pushedFilters;
}
@@ -214,6 +242,11 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
}
@Override
+ public Scan build() {
+ return this;
+ }
+
+ @Override
public Statistics estimateStatistics() {
long sizeInBytes = 0L;
long numRows = 0L;
@@ -228,57 +261,7 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
return new Stats(sizeInBytes, numRows);
}
- private List<CombinedScanTask> tasks() {
- if (tasks == null) {
- TableScan scan = table
- .newScan()
- .caseSensitive(caseSensitive)
- .project(lazySchema());
-
- if (snapshotId != null) {
- scan = scan.useSnapshot(snapshotId);
- }
-
- if (asOfTimestamp != null) {
- scan = scan.asOfTime(asOfTimestamp);
- }
-
- if (splitSize != null) {
- scan = scan.option(TableProperties.SPLIT_SIZE, splitSize.toString());
- }
-
- if (splitLookback != null) {
- scan = scan.option(TableProperties.SPLIT_LOOKBACK, splitLookback.toString());
- }
-
- if (splitOpenFileCost != null) {
- scan = scan.option(TableProperties.SPLIT_OPEN_FILE_COST, splitOpenFileCost.toString());
- }
-
- if (filterExpressions != null) {
- for (Expression filter : filterExpressions) {
- scan = scan.filter(filter);
- }
- }
-
- try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
- this.tasks = Lists.newArrayList(tasksIterable);
- } catch (IOException e) {
- throw new RuntimeIOException(e, "Failed to close table scan: %s", scan);
- }
- }
-
- return tasks;
- }
-
- @Override
- public String toString() {
- return String.format(
- "IcebergScan(table=%s, type=%s, filters=%s, caseSensitive=%s)",
- table, lazySchema().asStruct(), filterExpressions, caseSensitive);
- }
-
- private static class ReadTask implements InputPartition<InternalRow>, Serializable {
+ public static class BatchReadInputPartition implements InputPartition, Serializable {
private final CombinedScanTask task;
private final String tableSchemaString;
private final String expectedSchemaString;
@@ -289,9 +272,13 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private transient Schema tableSchema = null;
private transient Schema expectedSchema = null;
- private ReadTask(
- CombinedScanTask task, String tableSchemaString, String expectedSchemaString, FileIO fileIo,
- EncryptionManager encryptionManager, boolean caseSensitive) {
+ BatchReadInputPartition(
+ CombinedScanTask task,
+ String tableSchemaString,
+ String expectedSchemaString,
+ FileIO fileIo,
+ EncryptionManager encryptionManager,
+ boolean caseSensitive) {
this.task = task;
this.tableSchemaString = tableSchemaString;
this.expectedSchemaString = expectedSchemaString;
@@ -300,12 +287,6 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
this.caseSensitive = caseSensitive;
}
- @Override
- public InputPartitionReader<InternalRow> createPartitionReader() {
- return new TaskDataReader(task, lazyTableSchema(), lazyExpectedSchema(), fileIo,
- encryptionManager, caseSensitive);
- }
-
private Schema lazyTableSchema() {
if (tableSchema == null) {
this.tableSchema = SchemaParser.fromJson(tableSchemaString);
@@ -321,7 +302,160 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
}
}
- private static class TaskDataReader implements InputPartitionReader<InternalRow> {
+ @Override
+ public InputPartition[] planInputPartitions() {
+ String tableSchemaString = SchemaParser.toJson(table.schema());
+ String expectedSchemaString = SchemaParser.toJson(lazySchema());
+
+ List<CombinedScanTask> scanTasks = tasks();
+ InputPartition[] readTasks = new InputPartition[scanTasks.size()];
+ for (int i = 0; i < scanTasks.size(); i++) {
+ readTasks[i] = new BatchReadInputPartition(scanTasks.get(i), tableSchemaString, expectedSchemaString, fileIo,
+ encryptionManager, caseSensitive);
+ }
+
+ return readTasks;
+ }
+
+ @Override
+ public IcebergRowReaderFactory createReaderFactory() {
+ return new IcebergRowReaderFactory();
+ }
+
+ private Schema lazySchema() {
+ if (schema == null) {
+ if (requestedSchema != null) {
+ this.schema = SparkSchemaUtil.prune(table.schema(), requestedSchema);
+ } else {
+ this.schema = table.schema();
+ }
+ }
+ return schema;
+ }
+
+ private StructType lazyType() {
+ if (type == null) {
+ this.type = SparkSchemaUtil.convert(lazySchema());
+ }
+ return type;
+ }
+
+ @Override
+ public StructType readSchema() {
+ return lazyType();
+ }
+
+
+ public static class PartitionRowConverter implements Function<StructLike, InternalRow> {
+ private final DataType[] types;
+ private final int[] positions;
+ private final Class<?>[] javaTypes;
+ private final GenericInternalRow reusedRow;
+
+ PartitionRowConverter(Schema partitionSchema, PartitionSpec spec) {
+ StructType partitionType = SparkSchemaUtil.convert(partitionSchema);
+ StructField[] fields = partitionType.fields();
+
+ this.types = new DataType[fields.length];
+ this.positions = new int[types.length];
+ this.javaTypes = new Class<?>[types.length];
+ this.reusedRow = new GenericInternalRow(types.length);
+
+ List<PartitionField> partitionFields = spec.fields();
+ for (int rowIndex = 0; rowIndex < fields.length; rowIndex += 1) {
+ this.types[rowIndex] = fields[rowIndex].dataType();
+
+ int sourceId = partitionSchema.columns().get(rowIndex).fieldId();
+ for (int specIndex = 0; specIndex < partitionFields.size(); specIndex += 1) {
+ PartitionField field = spec.fields().get(specIndex);
+ if (field.sourceId() == sourceId && "identity".equals(field.transform().toString())) {
+ positions[rowIndex] = specIndex;
+ javaTypes[rowIndex] = spec.javaClasses()[specIndex];
+ break;
+ }
+ }
+ }
+ }
+
+ @Override
+ public InternalRow apply(StructLike tuple) {
+ for (int i = 0; i < types.length; i += 1) {
+ Object value = tuple.get(positions[i], javaTypes[i]);
+ if (value != null) {
+ reusedRow.update(i, convert(value, types[i]));
+ } else {
+ reusedRow.setNullAt(i);
+ }
+ }
+
+ return reusedRow;
+ }
+
+ /**
+ * Converts the objects into instances used by Spark's InternalRow.
+ *
+ * @param value a data value
+ * @param type the Spark data type
+ * @return the value converted to the representation expected by Spark's InternalRow.
+ */
+ private static Object convert(Object value, DataType type) {
+ if (type instanceof StringType) {
+ return UTF8String.fromString(value.toString());
+ } else if (type instanceof BinaryType) {
+ return ByteBuffers.toByteArray((ByteBuffer) value);
+ } else if (type instanceof DecimalType) {
+ return Decimal.fromDecimal(value);
+ }
+ return value;
+ }
+ }
+
+ public static class StructLikeInternalRow implements StructLike {
+ private final DataType[] types;
+ private InternalRow row = null;
+
+ StructLikeInternalRow(StructType struct) {
+ this.types = new DataType[struct.size()];
+ StructField[] fields = struct.fields();
+ for (int i = 0; i < fields.length; i += 1) {
+ types[i] = fields[i].dataType();
+ }
+ }
+
+ public StructLikeInternalRow setRow(InternalRow row) {
+ this.row = row;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return types.length;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(row.get(pos, types[pos]));
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Not implemented: set");
+ }
+ }
+
+
+ private static class IcebergRowReaderFactory implements PartitionReaderFactory {
+ IcebergRowReaderFactory() {
+ }
+
+ @Override
+ public PartitionReader<InternalRow> createReader(InputPartition inputPartition) {
+ return new TaskDataReader(inputPartition);
+ }
+ }
+
+ public static class TaskDataReader implements PartitionReader<InternalRow> {
// for some reason, the apply method can't be called from Java without reflection
private static final DynMethods.UnboundMethod APPLY_PROJECTION = DynMethods.builder("apply")
.impl(UnsafeProjection.class, InternalRow.class)
@@ -338,23 +472,25 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private Closeable currentCloseable = null;
private InternalRow current = null;
- TaskDataReader(CombinedScanTask task, Schema tableSchema, Schema expectedSchema, FileIO fileIo,
- EncryptionManager encryptionManager, boolean caseSensitive) {
- this.fileIo = fileIo;
- this.tasks = task.files().iterator();
- this.tableSchema = tableSchema;
- this.expectedSchema = expectedSchema;
- Iterable<InputFile> decryptedFiles = encryptionManager.decrypt(Iterables.transform(task.files(),
- fileScanTask ->
- EncryptedFiles.encryptedInput(
- this.fileIo.newInputFile(fileScanTask.file().path().toString()),
- fileScanTask.file().keyMetadata())));
+ TaskDataReader(InputPartition inputPartition) {
+ BatchReadInputPartition batchReadInputPartition = (BatchReadInputPartition) inputPartition;
+
+ this.fileIo = batchReadInputPartition.fileIo;
+ this.tableSchema = batchReadInputPartition.lazyTableSchema();
+ this.expectedSchema = batchReadInputPartition.lazyExpectedSchema();
+ this.tasks = batchReadInputPartition.task.files().iterator();
+ Iterable<InputFile> decryptedFiles = batchReadInputPartition.encryptionManager.decrypt(
+ Iterables.transform(batchReadInputPartition.task.files(),
+ fileScanTask ->
+ EncryptedFiles.encryptedInput(
+ this.fileIo.newInputFile(fileScanTask.file().path().toString()),
+ fileScanTask.file().keyMetadata())));
ImmutableMap.Builder<String, InputFile> inputFileBuilder = ImmutableMap.builder();
decryptedFiles.forEach(decrypted -> inputFileBuilder.put(decrypted.location(), decrypted));
this.inputFiles = inputFileBuilder.build();
// open last because the schemas and fileIo must be set
this.currentIterator = open(tasks.next());
- this.caseSensitive = caseSensitive;
+ this.caseSensitive = batchReadInputPartition.caseSensitive;
}
@Override
@@ -535,101 +671,4 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
}
}
- private static class PartitionRowConverter implements Function<StructLike, InternalRow> {
- private final DataType[] types;
- private final int[] positions;
- private final Class<?>[] javaTypes;
- private final GenericInternalRow reusedRow;
-
- PartitionRowConverter(Schema partitionSchema, PartitionSpec spec) {
- StructType partitionType = SparkSchemaUtil.convert(partitionSchema);
- StructField[] fields = partitionType.fields();
-
- this.types = new DataType[fields.length];
- this.positions = new int[types.length];
- this.javaTypes = new Class<?>[types.length];
- this.reusedRow = new GenericInternalRow(types.length);
-
- List<PartitionField> partitionFields = spec.fields();
- for (int rowIndex = 0; rowIndex < fields.length; rowIndex += 1) {
- this.types[rowIndex] = fields[rowIndex].dataType();
-
- int sourceId = partitionSchema.columns().get(rowIndex).fieldId();
- for (int specIndex = 0; specIndex < partitionFields.size(); specIndex += 1) {
- PartitionField field = spec.fields().get(specIndex);
- if (field.sourceId() == sourceId && "identity".equals(field.transform().toString())) {
- positions[rowIndex] = specIndex;
- javaTypes[rowIndex] = spec.javaClasses()[specIndex];
- break;
- }
- }
- }
- }
-
- @Override
- public InternalRow apply(StructLike tuple) {
- for (int i = 0; i < types.length; i += 1) {
- Object value = tuple.get(positions[i], javaTypes[i]);
- if (value != null) {
- reusedRow.update(i, convert(value, types[i]));
- } else {
- reusedRow.setNullAt(i);
- }
- }
-
- return reusedRow;
- }
-
- /**
- * Converts the objects into instances used by Spark's InternalRow.
- *
- * @param value a data value
- * @param type the Spark data type
- * @return the value converted to the representation expected by Spark's InternalRow.
- */
- private static Object convert(Object value, DataType type) {
- if (type instanceof StringType) {
- return UTF8String.fromString(value.toString());
- } else if (type instanceof BinaryType) {
- return ByteBuffers.toByteArray((ByteBuffer) value);
- } else if (type instanceof DecimalType) {
- return Decimal.fromDecimal(value);
- }
- return value;
- }
- }
-
- private static class StructLikeInternalRow implements StructLike {
- private final DataType[] types;
- private InternalRow row = null;
-
- StructLikeInternalRow(StructType struct) {
- this.types = new DataType[struct.size()];
- StructField[] fields = struct.fields();
- for (int i = 0; i < fields.length; i += 1) {
- types[i] = fields[i].dataType();
- }
- }
-
- public StructLikeInternalRow setRow(InternalRow row) {
- this.row = row;
- return this;
- }
-
- @Override
- public int size() {
- return types.length;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public <T> T get(int pos, Class<T> javaClass) {
- return javaClass.cast(row.get(pos, types[pos]));
- }
-
- @Override
- public <T> void set(int pos, T value) {
- throw new UnsupportedOperationException("Not implemented: set");
- }
- }
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java
similarity index 84%
rename from spark/src/main/java/org/apache/iceberg/spark/source/Writer.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java
index f856a5a..dfdf6b1 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergBatchWriter.java
@@ -37,6 +37,7 @@ import org.apache.iceberg.DataFiles;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.Schema;
@@ -47,6 +48,7 @@ import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
@@ -57,11 +59,12 @@ import org.apache.iceberg.spark.data.SparkParquetWriters;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.iceberg.util.Tasks;
import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.writer.DataSourceWriter;
-import org.apache.spark.sql.sources.v2.writer.DataWriter;
-import org.apache.spark.sql.sources.v2.writer.DataWriterFactory;
-import org.apache.spark.sql.sources.v2.writer.WriterCommitMessage;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.write.BatchWrite;
+import org.apache.spark.sql.connector.write.DataWriter;
+import org.apache.spark.sql.connector.write.DataWriterFactory;
+import org.apache.spark.sql.connector.write.WriterCommitMessage;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -78,31 +81,31 @@ import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
-// TODO: parameterize DataSourceWriter with subclass of WriterCommitMessage
-class Writer implements DataSourceWriter {
- private static final Logger LOG = LoggerFactory.getLogger(Writer.class);
+class IcebergBatchWriter implements BatchWrite {
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergBatchWriter.class);
private final Table table;
private final FileFormat format;
private final FileIO fileIo;
private final EncryptionManager encryptionManager;
- private final boolean replacePartitions;
+ private final TableCapability writeBehavior;
private final String applicationId;
private final String wapId;
private final long targetFileSize;
private final Schema dsSchema;
- Writer(Table table, DataSourceOptions options, boolean replacePartitions, String applicationId, Schema dsSchema) {
- this(table, options, replacePartitions, applicationId, null, dsSchema);
- }
-
- Writer(Table table, DataSourceOptions options, boolean replacePartitions, String applicationId, String wapId,
+ IcebergBatchWriter(
+ Table table,
+ CaseInsensitiveStringMap options,
+ TableCapability writeBehavior,
+ String applicationId,
+ String wapId,
Schema dsSchema) {
this.table = table;
this.format = getFileFormat(table.properties(), options);
this.fileIo = table.io();
this.encryptionManager = table.encryption();
- this.replacePartitions = replacePartitions;
+ this.writeBehavior = writeBehavior;
this.applicationId = applicationId;
this.wapId = wapId;
this.dsSchema = dsSchema;
@@ -112,8 +115,8 @@ class Writer implements DataSourceWriter {
this.targetFileSize = options.getLong("target-file-size-bytes", tableTargetFileSize);
}
- private FileFormat getFileFormat(Map<String, String> tableProperties, DataSourceOptions options) {
- Optional<String> formatOption = options.get("write-format");
+ protected FileFormat getFileFormat(Map<String, String> tableProperties, Map<String, String> options) {
+ Optional<String> formatOption = Optional.ofNullable(options.get("write-format"));
String formatString = formatOption
.orElse(tableProperties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT));
return FileFormat.valueOf(formatString.toUpperCase(Locale.ENGLISH));
@@ -125,7 +128,7 @@ class Writer implements DataSourceWriter {
}
@Override
- public DataWriterFactory<InternalRow> createWriterFactory() {
+ public DataWriterFactory createBatchWriterFactory() {
return new WriterFactory(
table.spec(), format, table.locationProvider(), table.properties(), fileIo, encryptionManager, targetFileSize,
dsSchema);
@@ -133,10 +136,14 @@ class Writer implements DataSourceWriter {
@Override
public void commit(WriterCommitMessage[] messages) {
- if (replacePartitions) {
+ if (writeBehavior.equals(TableCapability.OVERWRITE_DYNAMIC)) {
replacePartitions(messages);
- } else {
+ } else if (writeBehavior.equals(TableCapability.BATCH_WRITE)) {
append(messages);
+ } else if (writeBehavior.equals(TableCapability.TRUNCATE)) {
+ overwrite(messages);
+ } else {
+ throw new IllegalArgumentException("Iceberg doen't support write behavior " + writeBehavior + " for now");
}
}
@@ -183,6 +190,19 @@ class Writer implements DataSourceWriter {
commitOperation(dynamicOverwrite, numFiles, "dynamic partition overwrite");
}
+ private void overwrite(WriterCommitMessage[] messages) {
+ OverwriteFiles overwriteFiles = table.newOverwrite();
+ overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
+
+ int numFiles = 0;
+ for (DataFile file : files(messages)) {
+ numFiles += 1;
+ overwriteFiles.addFile(file);
+ }
+
+ commitOperation(overwriteFiles, numFiles, "overwrite by filter or truncate");
+ }
+
@Override
public void abort(WriterCommitMessage[] messages) {
Tasks.foreach(files(messages))
@@ -246,7 +266,7 @@ class Writer implements DataSourceWriter {
}
}
- private static class WriterFactory implements DataWriterFactory<InternalRow> {
+ protected static class WriterFactory implements DataWriterFactory {
private final PartitionSpec spec;
private final FileFormat format;
private final LocationProvider locations;
@@ -256,9 +276,15 @@ class Writer implements DataSourceWriter {
private final long targetFileSize;
private final Schema dsSchema;
- WriterFactory(PartitionSpec spec, FileFormat format, LocationProvider locations,
- Map<String, String> properties, FileIO fileIo, EncryptionManager encryptionManager,
- long targetFileSize, Schema dsSchema) {
+ WriterFactory(
+ PartitionSpec spec,
+ FileFormat format,
+ LocationProvider locations,
+ Map<String, String> properties,
+ FileIO fileIo,
+ EncryptionManager encryptionManager,
+ long targetFileSize,
+ Schema dsSchema) {
this.spec = spec;
this.format = format;
this.locations = locations;
@@ -269,10 +295,19 @@ class Writer implements DataSourceWriter {
this.dsSchema = dsSchema;
}
- @Override
- public DataWriter<InternalRow> createDataWriter(int partitionId, long taskId, long epochId) {
+ public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
OutputFileFactory fileFactory = new OutputFileFactory(partitionId, taskId, epochId);
AppenderFactory<InternalRow> appenderFactory = new SparkAppenderFactory();
+ if (spec.fields().isEmpty()) {
+ return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
+ } else {
+ return new PartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
+ }
+ }
+
+ public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
+ OutputFileFactory fileFactory = new OutputFileFactory(partitionId, taskId, 0);
+ AppenderFactory<InternalRow> appenderFactory = new SparkAppenderFactory();
if (spec.fields().isEmpty()) {
return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
@@ -371,8 +406,13 @@ class Writer implements DataSourceWriter {
private EncryptedOutputFile currentFile = null;
private long currentRows = 0;
- BaseWriter(PartitionSpec spec, FileFormat format, AppenderFactory<InternalRow> appenderFactory,
- WriterFactory.OutputFileFactory fileFactory, FileIO fileIo, long targetFileSize) {
+ BaseWriter(
+ PartitionSpec spec,
+ FileFormat format,
+ AppenderFactory<InternalRow> appenderFactory,
+ WriterFactory.OutputFileFactory fileFactory,
+ FileIO fileIo,
+ long targetFileSize) {
this.spec = spec;
this.format = format;
this.appenderFactory = appenderFactory;
@@ -384,7 +424,7 @@ class Writer implements DataSourceWriter {
@Override
public abstract void write(InternalRow row) throws IOException;
- public void writeInternal(InternalRow row) throws IOException {
+ public void writeInternal(InternalRow row) throws IOException {
if (currentRows % ROWS_DIVISOR == 0 && currentAppender.length() >= targetFileSize) {
closeCurrent();
openCurrent();
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
deleted file mode 100644
index f97de81..0000000
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
+++ /dev/null
@@ -1,204 +0,0 @@
-/*
- * 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.iceberg.spark.source;
-
-import com.google.common.base.Preconditions;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Collectors;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.hadoop.HadoopTables;
-import org.apache.iceberg.hive.HiveCatalog;
-import org.apache.iceberg.hive.HiveCatalogs;
-import org.apache.iceberg.spark.SparkSchemaUtil;
-import org.apache.iceberg.transforms.Transform;
-import org.apache.iceberg.transforms.UnknownTransform;
-import org.apache.iceberg.types.CheckCompatibility;
-import org.apache.spark.sql.SaveMode;
-import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.execution.streaming.StreamExecution;
-import org.apache.spark.sql.sources.DataSourceRegister;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.DataSourceV2;
-import org.apache.spark.sql.sources.v2.ReadSupport;
-import org.apache.spark.sql.sources.v2.StreamWriteSupport;
-import org.apache.spark.sql.sources.v2.WriteSupport;
-import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
-import org.apache.spark.sql.sources.v2.writer.DataSourceWriter;
-import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
-import org.apache.spark.sql.streaming.OutputMode;
-import org.apache.spark.sql.types.StructType;
-
-public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, DataSourceRegister, StreamWriteSupport {
-
- private SparkSession lazySpark = null;
- private Configuration lazyConf = null;
-
- @Override
- public String shortName() {
- return "iceberg";
- }
-
- @Override
- public DataSourceReader createReader(DataSourceOptions options) {
- return createReader(null, options);
- }
-
- @Override
- public DataSourceReader createReader(StructType readSchema, DataSourceOptions options) {
- Configuration conf = new Configuration(lazyBaseConf());
- Table table = getTableAndResolveHadoopConfiguration(options, conf);
- String caseSensitive = lazySparkSession().conf().get("spark.sql.caseSensitive");
-
- Reader reader = new Reader(table, Boolean.parseBoolean(caseSensitive), options);
- if (readSchema != null) {
- // convert() will fail if readSchema contains fields not in table.schema()
- SparkSchemaUtil.convert(table.schema(), readSchema);
- reader.pruneColumns(readSchema);
- }
-
- return reader;
- }
-
- @Override
- public Optional<DataSourceWriter> createWriter(String jobId, StructType dsStruct, SaveMode mode,
- DataSourceOptions options) {
- Preconditions.checkArgument(mode == SaveMode.Append || mode == SaveMode.Overwrite,
- "Save mode %s is not supported", mode);
- Configuration conf = new Configuration(lazyBaseConf());
- Table table = getTableAndResolveHadoopConfiguration(options, conf);
- Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
- validateWriteSchema(table.schema(), dsSchema, checkNullability(options));
- validatePartitionTransforms(table.spec());
- String appId = lazySparkSession().sparkContext().applicationId();
- String wapId = lazySparkSession().conf().get("spark.wap.id", null);
- return Optional.of(new Writer(table, options, mode == SaveMode.Overwrite, appId, wapId, dsSchema));
- }
-
- @Override
- public StreamWriter createStreamWriter(String runId, StructType dsStruct,
- OutputMode mode, DataSourceOptions options) {
- Preconditions.checkArgument(
- mode == OutputMode.Append() || mode == OutputMode.Complete(),
- "Output mode %s is not supported", mode);
- Configuration conf = new Configuration(lazyBaseConf());
- Table table = getTableAndResolveHadoopConfiguration(options, conf);
- Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
- validateWriteSchema(table.schema(), dsSchema, checkNullability(options));
- validatePartitionTransforms(table.spec());
- // Spark 2.4.x passes runId to createStreamWriter instead of real queryId,
- // so we fetch it directly from sparkContext to make writes idempotent
- String queryId = lazySparkSession().sparkContext().getLocalProperty(StreamExecution.QUERY_ID_KEY());
- String appId = lazySparkSession().sparkContext().applicationId();
- return new StreamingWriter(table, options, queryId, mode, appId, dsSchema);
- }
-
- protected Table findTable(DataSourceOptions options, Configuration conf) {
- Optional<String> path = options.get("path");
- Preconditions.checkArgument(path.isPresent(), "Cannot open table: path is not set");
-
- if (path.get().contains("/")) {
- HadoopTables tables = new HadoopTables(conf);
- return tables.load(path.get());
- } else {
- HiveCatalog hiveCatalog = HiveCatalogs.loadCatalog(conf);
- TableIdentifier tableIdentifier = TableIdentifier.parse(path.get());
- return hiveCatalog.loadTable(tableIdentifier);
- }
- }
-
- private SparkSession lazySparkSession() {
- if (lazySpark == null) {
- this.lazySpark = SparkSession.builder().getOrCreate();
- }
- return lazySpark;
- }
-
- private Configuration lazyBaseConf() {
- if (lazyConf == null) {
- this.lazyConf = lazySparkSession().sessionState().newHadoopConf();
- }
- return lazyConf;
- }
-
- private Table getTableAndResolveHadoopConfiguration(
- DataSourceOptions options, Configuration conf) {
- // Overwrite configurations from the Spark Context with configurations from the options.
- mergeIcebergHadoopConfs(conf, options.asMap());
- Table table = findTable(options, conf);
- // Set confs from table properties
- mergeIcebergHadoopConfs(conf, table.properties());
- // Re-overwrite values set in options and table properties but were not in the environment.
- mergeIcebergHadoopConfs(conf, options.asMap());
- return table;
- }
-
- private static void mergeIcebergHadoopConfs(
- Configuration baseConf, Map<String, String> options) {
- options.keySet().stream()
- .filter(key -> key.startsWith("hadoop."))
- .forEach(key -> baseConf.set(key.replaceFirst("hadoop.", ""), options.get(key)));
- }
-
- private void validateWriteSchema(Schema tableSchema, Schema dsSchema, Boolean checkNullability) {
- List<String> errors;
- if (checkNullability) {
- errors = CheckCompatibility.writeCompatibilityErrors(tableSchema, dsSchema);
- } else {
- errors = CheckCompatibility.typeCompatibilityErrors(tableSchema, dsSchema);
- }
- if (!errors.isEmpty()) {
- StringBuilder sb = new StringBuilder();
- sb.append("Cannot write incompatible dataset to table with schema:\n")
- .append(tableSchema)
- .append("\nProblems:");
- for (String error : errors) {
- sb.append("\n* ").append(error);
- }
- throw new IllegalArgumentException(sb.toString());
- }
- }
-
- private void validatePartitionTransforms(PartitionSpec spec) {
- if (spec.fields().stream().anyMatch(field -> field.transform() instanceof UnknownTransform)) {
- String unsupported = spec.fields().stream()
- .map(PartitionField::transform)
- .filter(transform -> transform instanceof UnknownTransform)
- .map(Transform::toString)
- .collect(Collectors.joining(", "));
-
- throw new UnsupportedOperationException(
- String.format("Cannot write using unsupported transforms: %s", unsupported));
- }
- }
-
- private boolean checkNullability(DataSourceOptions options) {
- boolean sparkCheckNullability = Boolean.parseBoolean(lazySpark.conf()
- .get("spark.sql.iceberg.check-nullability", "true"));
- boolean dataFrameCheckNullability = options.getBoolean("check-nullability", true);
- return sparkCheckNullability && dataFrameCheckNullability;
- }
-}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java
similarity index 53%
rename from spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java
rename to spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java
index fa3cbd0..e1d0dfb 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergStreamingWriter.java
@@ -22,38 +22,80 @@ package org.apache.iceberg.spark.source;
import java.util.Map;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotUpdate;
import org.apache.iceberg.Table;
+import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.expressions.Expressions;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.writer.WriterCommitMessage;
-import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
-import org.apache.spark.sql.streaming.OutputMode;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.write.DataWriter;
+import org.apache.spark.sql.connector.write.WriterCommitMessage;
+import org.apache.spark.sql.connector.write.streaming.StreamingDataWriterFactory;
+import org.apache.spark.sql.connector.write.streaming.StreamingWrite;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class StreamingWriter extends Writer implements StreamWriter {
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
+import static org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT;
- private static final Logger LOG = LoggerFactory.getLogger(StreamingWriter.class);
+public class IcebergStreamingWriter extends IcebergBatchWriter implements StreamingWrite {
+
+ private static final Logger LOG = LoggerFactory.getLogger(IcebergStreamingWriter.class);
private static final String QUERY_ID_PROPERTY = "spark.sql.streaming.queryId";
private static final String EPOCH_ID_PROPERTY = "spark.sql.streaming.epochId";
private final String queryId;
- private final OutputMode mode;
+ private final TableCapability writeBehavior;
+ private final Table table;
+ private final long targetFileSize;
+ private final FileFormat format;
+ private final Schema dsSchema;
- StreamingWriter(Table table, DataSourceOptions options, String queryId, OutputMode mode, String applicationId,
- Schema dsSchema) {
- super(table, options, false, applicationId, dsSchema);
+ IcebergStreamingWriter(Table table, CaseInsensitiveStringMap options, String queryId, TableCapability writeBehavior,
+ String applicationId, String wapId, Schema dsSchema) {
+ super(table, options, writeBehavior, applicationId, wapId, dsSchema);
this.queryId = queryId;
- this.mode = mode;
+ this.writeBehavior = writeBehavior;
+ this.table = table;
+ this.format = getFileFormat(table.properties(), options);
+ long tableTargetFileSize = PropertyUtil.propertyAsLong(
+ table.properties(), WRITE_TARGET_FILE_SIZE_BYTES, WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
+ this.targetFileSize = options.getLong("target-file-size-bytes", tableTargetFileSize);
+ this.dsSchema = dsSchema;
+ }
+
+ @Override
+ public StreamingDataWriterFactory createStreamingWriterFactory() {
+ return new StreamingWriterFactory(table.spec(), format, table.locationProvider(),
+ table.properties(), table.io(), table.encryption(), targetFileSize, dsSchema);
+ }
+
+ private static class StreamingWriterFactory extends WriterFactory implements StreamingDataWriterFactory {
+
+ StreamingWriterFactory(PartitionSpec spec, FileFormat format, LocationProvider locations,
+ Map<String, String> properties, FileIO fileIo, EncryptionManager encryptionManager,
+ long targetFileSize, Schema dsSchema) {
+ super(spec, format, locations, properties, fileIo, encryptionManager, targetFileSize, dsSchema);
+ }
+
+ @Override
+ public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
+ return super.createWriter(partitionId, taskId, epochId);
+ }
}
@Override
public void commit(long epochId, WriterCommitMessage[] messages) {
- LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, mode);
+ LOG.info("Committing epoch {} for query {} in {} mode", epochId, queryId, writeBehavior);
table().refresh();
Long lastCommittedEpochId = getLastCommittedEpochId();
@@ -62,7 +104,7 @@ public class StreamingWriter extends Writer implements StreamWriter {
return;
}
- if (mode == OutputMode.Complete()) {
+ if (writeBehavior.equals(TableCapability.TRUNCATE)) {
OverwriteFiles overwriteFiles = table().newOverwrite();
overwriteFiles.overwriteByRowFilter(Expressions.alwaysTrue());
int numFiles = 0;
@@ -71,7 +113,7 @@ public class StreamingWriter extends Writer implements StreamWriter {
numFiles++;
}
commit(overwriteFiles, epochId, numFiles, "streaming complete overwrite");
- } else {
+ } else if (writeBehavior.equals(TableCapability.STREAMING_WRITE)) {
AppendFiles append = table().newFastAppend();
int numFiles = 0;
for (DataFile file : files(messages)) {
@@ -79,6 +121,8 @@ public class StreamingWriter extends Writer implements StreamWriter {
numFiles++;
}
commit(append, epochId, numFiles, "streaming append");
+ } else {
+ throw new IllegalArgumentException("Iceberg doen't support write behavior " + writeBehavior + " for now");
}
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java
new file mode 100644
index 0000000..d02645d
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTable.java
@@ -0,0 +1,222 @@
+/*
+ * 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.iceberg.spark.source;
+
+import com.google.common.collect.ImmutableSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.spark.SparkUtils;
+import org.apache.iceberg.transforms.UnknownTransform;
+import org.apache.iceberg.types.CheckCompatibility;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+import org.apache.spark.sql.connector.catalog.SupportsWrite;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.connector.read.ScanBuilder;
+import org.apache.spark.sql.connector.write.BatchWrite;
+import org.apache.spark.sql.connector.write.SupportsDynamicOverwrite;
+import org.apache.spark.sql.connector.write.SupportsTruncate;
+import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.connector.write.streaming.StreamingWrite;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+final class IcebergTable implements Table, SupportsRead, SupportsWrite {
+
+ private static final Set<TableCapability> CAPABILITIES = ImmutableSet.of(
+ TableCapability.BATCH_READ,
+ TableCapability.BATCH_WRITE,
+ TableCapability.MICRO_BATCH_READ,
+ TableCapability.STREAMING_WRITE,
+ TableCapability.TRUNCATE,
+ TableCapability.OVERWRITE_DYNAMIC);
+
+ private final org.apache.iceberg.Table tableInIceberg;
+ private StructType requestSchema;
+
+ IcebergTable(org.apache.iceberg.Table tableInIceberg, StructType requestSchema) {
+ this.tableInIceberg = tableInIceberg;
+
+ if (requestSchema != null) {
+ SparkSchemaUtil.convert(tableInIceberg.schema(), requestSchema);
+ this.requestSchema = requestSchema;
+ }
+ }
+
+ @Override
+ public String name() {
+ return tableInIceberg.name();
+ }
+
+ @Override
+ public StructType schema() {
+ if (requestSchema != null) {
+ return requestSchema;
+ }
+ return SparkSchemaUtil.convert(tableInIceberg.schema());
+ }
+
+ @Override
+ public Transform[] partitioning() {
+ return SparkUtils.toTransforms(tableInIceberg.spec());
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return tableInIceberg.properties();
+ }
+
+ @Override
+ public Set<TableCapability> capabilities() {
+ return CAPABILITIES;
+ }
+
+ @Override
+ public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
+ return () -> new IcebergBatchScan(tableInIceberg, options, requestSchema);
+ }
+
+ @Override
+ public WriteBuilder newWriteBuilder(CaseInsensitiveStringMap options) {
+ return new IcebergWriteBuilder(tableInIceberg, options);
+ }
+
+ static class IcebergWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsTruncate {
+
+ private org.apache.iceberg.Table table;
+ private CaseInsensitiveStringMap writeOptions;
+ private TableCapability writeBehavior = TableCapability.BATCH_WRITE;
+ private String writeQueryId = null;
+ private StructType dsStruct = null;
+
+ IcebergWriteBuilder(org.apache.iceberg.Table table, CaseInsensitiveStringMap options) {
+ this.table = table;
+ this.writeOptions = options;
+ }
+
+ @Override
+ public WriteBuilder withQueryId(String queryId) {
+ this.writeQueryId = queryId;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder withInputDataSchema(StructType schemaInput) {
+ this.dsStruct = schemaInput;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder overwriteDynamicPartitions() {
+ this.writeBehavior = TableCapability.OVERWRITE_DYNAMIC;
+ return this;
+ }
+
+ @Override
+ public WriteBuilder truncate() {
+ this.writeBehavior = TableCapability.TRUNCATE;
+ return this;
+ }
+
+ @Override
+ public BatchWrite buildForBatch() {
+ // TODO. Check queryId and schema before build?
+
+ // Validate
+ Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
+ validateWriteSchema(table.schema(), dsSchema, checkNullability(writeOptions));
+ validatePartitionTransforms(table.spec());
+
+ // Get application id
+ String appId = SparkUtils.getSparkSession().sparkContext().applicationId();
+
+ // Get write-audit-publish id
+ String wapId = SparkUtils.getSparkSession().conf().get("spark.wap.id", null);
+
+ return new IcebergBatchWriter(table, writeOptions, writeBehavior, appId, wapId, dsSchema);
+ }
+
+ @Override
+ public StreamingWrite buildForStreaming() {
+ // TODO. Check queryId and schema before build?
+
+ // Validate
+ Schema dsSchema = SparkSchemaUtil.convert(table.schema(), dsStruct);
+ validateWriteSchema(table.schema(), dsSchema, checkNullability(writeOptions));
+ validatePartitionTransforms(table.spec());
+
+ // Change to streaming write if it is just append
+ if (writeBehavior.equals(TableCapability.BATCH_WRITE)) {
+ writeBehavior = TableCapability.STREAMING_WRITE;
+ }
+
+ // Get application id
+ String appId = SparkUtils.getSparkSession().sparkContext().applicationId();
+ String wapId = SparkUtils.getSparkSession().conf().get("spark.wap.id", null);
+ return new IcebergStreamingWriter(table, writeOptions, writeQueryId, writeBehavior, appId, wapId, table.schema());
+ }
+
+ private void validateWriteSchema(Schema tableSchema, Schema dsSchema, Boolean checkNullability) {
+ List<String> errors;
+ if (checkNullability) {
+ errors = CheckCompatibility.writeCompatibilityErrors(tableSchema, dsSchema);
+ } else {
+ errors = CheckCompatibility.typeCompatibilityErrors(tableSchema, dsSchema);
+ }
+ if (!errors.isEmpty()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Cannot write incompatible dataset to table with schema:\n")
+ .append(tableSchema)
+ .append("\nProblems:");
+ for (String error : errors) {
+ sb.append("\n* ").append(error);
+ }
+ throw new IllegalArgumentException(sb.toString());
+ }
+ }
+
+ private void validatePartitionTransforms(PartitionSpec spec) {
+ if (spec.fields().stream().anyMatch(field -> field.transform() instanceof UnknownTransform)) {
+ String unsupported = spec.fields().stream()
+ .map(PartitionField::transform)
+ .filter(transform -> transform instanceof UnknownTransform)
+ .map(org.apache.iceberg.transforms.Transform::toString)
+ .collect(Collectors.joining(", "));
+
+ throw new UnsupportedOperationException(
+ String.format("Cannot write using unsupported transforms: %s", unsupported));
+ }
+ }
+
+ private boolean checkNullability(CaseInsensitiveStringMap options) {
+ boolean sparkCheckNullability = Boolean.parseBoolean(SparkUtils.getSparkSession().conf()
+ .get("spark.sql.iceberg.check-nullability", "true"));
+ boolean dataFrameCheckNullability = options.getBoolean("check-nullability", true);
+ return sparkCheckNullability && dataFrameCheckNullability;
+ }
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java
new file mode 100644
index 0000000..9f7e1d9
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergTableProvider.java
@@ -0,0 +1,101 @@
+/*
+ * 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.iceberg.spark.source;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.spark.SparkUtils;
+import org.apache.spark.sql.connector.catalog.Table;
+import org.apache.spark.sql.connector.catalog.TableProvider;
+import org.apache.spark.sql.sources.DataSourceRegister;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+public class IcebergTableProvider implements DataSourceRegister, TableProvider {
+ @Override
+ public String shortName() {
+ return "iceberg";
+ }
+
+ @Override
+ public Table getTable(CaseInsensitiveStringMap options) {
+ return getTable(options, null);
+ }
+
+ @Override
+ public Table getTable(CaseInsensitiveStringMap options, StructType readSchema) {
+ // Get Iceberg table from options
+ Configuration conf = new Configuration(SparkUtils.getBaseConf());
+ org.apache.iceberg.Table tableInIceberg = getTableAndResolveHadoopConfiguration(options, conf);
+
+ // Build Spark table based on Iceberg table, and return it
+ return new IcebergTable(tableInIceberg, readSchema);
+ }
+
+ protected org.apache.iceberg.Table getTableAndResolveHadoopConfiguration(
+ CaseInsensitiveStringMap options, Configuration conf) {
+ // Overwrite configurations from the Spark Context with configurations from the options.
+ mergeIcebergHadoopConfs(conf, options);
+
+ // Find table (in Iceberg) based on the given path
+ org.apache.iceberg.Table table = findTable(options, conf);
+
+ // Set confs from table properties
+ mergeIcebergHadoopConfs(conf, table.properties());
+
+ // Re-overwrite values set in options and table properties but were not in the environment.
+ mergeIcebergHadoopConfs(conf, options);
+
+ return table;
+ }
+
+ /**
+ * Merge delta options into base conf
+ *
+ * @param baseConf the base conf
+ * @param options the delta options to merge into base
+ */
+ private void mergeIcebergHadoopConfs(Configuration baseConf, Map<String, String> options) {
+ options.keySet().stream()
+ .filter(key -> key.startsWith("hadoop.")) /* filter all keys staring with "hadoop." */
+ .forEach(key -> baseConf.set(key.replaceFirst("hadoop.", ""), options.get(key)));
+ /* Modify the key by removing the prefix of "hadoop." and merge into base */
+ }
+
+ protected org.apache.iceberg.Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
+ Optional<String> path = Optional.ofNullable(options.get("path"));
+ Preconditions.checkArgument(path.isPresent(), "Cannot open table: path is not set");
+
+ if (path.get().contains("/")) { // hadoop table
+ HadoopTables tables = new HadoopTables(conf);
+ return tables.load(path.get());
+ } else { // hive table
+ HiveCatalog hiveCatalog = HiveCatalogs.loadCatalog(conf);
+ TableIdentifier tableIdentifier = TableIdentifier.parse(path.get());
+ return hiveCatalog.loadTable(tableIdentifier);
+ }
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/Stats.java b/spark/src/main/java/org/apache/iceberg/spark/source/Stats.java
index 76119c1..939b07a 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/Stats.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/Stats.java
@@ -20,7 +20,7 @@
package org.apache.iceberg.spark.source;
import java.util.OptionalLong;
-import org.apache.spark.sql.sources.v2.reader.Statistics;
+import org.apache.spark.sql.connector.read.Statistics;
class Stats implements Statistics {
private final OptionalLong sizeInBytes;
diff --git a/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
index 01a6c4e..04e5f32 100644
--- a/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
+++ b/spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -17,4 +17,4 @@
# under the License.
#
-org.apache.iceberg.spark.source.IcebergSource
+org.apache.iceberg.spark.source.IcebergTableProvider
diff --git a/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkDateTimes.java b/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkDateTimes.java
index 10da4da..bbd8bbf 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkDateTimes.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkDateTimes.java
@@ -19,10 +19,12 @@
package org.apache.iceberg.spark.data;
+import java.time.ZoneId;
import java.util.TimeZone;
import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.catalyst.util.DateTimeUtils;
+import org.apache.spark.sql.catalyst.util.TimestampFormatter;
import org.junit.Assert;
import org.junit.Test;
@@ -64,8 +66,11 @@ public class TestSparkDateTimes {
}
public void checkSparkTimestamp(String timestampString, String sparkRepr) {
+ ZoneId zoneid = ZoneId.of("UTC");
+ TimestampFormatter timestampFormatter = TimestampFormatter.getFractionFormatter(zoneid);
+
Literal<Long> ts = Literal.of(timestampString).to(Types.TimestampType.withZone());
- String sparkTimestamp = DateTimeUtils.timestampToString(ts.value());
+ String sparkTimestamp = DateTimeUtils.timestampToString(timestampFormatter, ts.value());
System.err.println(timestampString + ": " + ts.value());
Assert.assertEquals("Should be the same timestamp (" + ts.value() + ")",
sparkRepr, sparkTimestamp);
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
index e189ab3..9ff583d 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestDataFrameWrites.java
@@ -54,6 +54,7 @@ import org.apache.spark.sql.catalyst.InternalRow;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -220,7 +221,8 @@ public class TestDataFrameWrites extends AvroDataTest {
return spark.internalCreateDataFrame(JavaRDD.toRDD(rdd), convert(schema), false);
}
- @Test
+ // This fails due to SPARK-28730
+ @Ignore
public void testNullableWithWriteOption() throws IOException {
File location = new File(temp.newFolder("parquet"), "test");
String sourcePath = String.format("%s/nullable_poc/sourceFolder/", location.toString());
@@ -264,7 +266,8 @@ public class TestDataFrameWrites extends AvroDataTest {
}
- @Test
+ // This fails due to SPARK-28730
+ @Ignore
public void testNullableWithSparkSqlOption() throws IOException {
File location = new File(temp.newFolder("parquet"), "test");
String sourcePath = String.format("%s/nullable_poc/sourceFolder/", location.toString());
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
index 389a11f..991e01d 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java
@@ -50,21 +50,20 @@ import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
-import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.UnsafeRow;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.SupportsPushDownFilters;
import org.apache.spark.sql.sources.And;
import org.apache.spark.sql.sources.EqualTo;
import org.apache.spark.sql.sources.Filter;
import org.apache.spark.sql.sources.GreaterThan;
import org.apache.spark.sql.sources.LessThan;
import org.apache.spark.sql.sources.StringStartsWith;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
-import org.apache.spark.sql.sources.v2.reader.InputPartition;
-import org.apache.spark.sql.sources.v2.reader.SupportsPushDownFilters;
import org.apache.spark.sql.types.IntegerType$;
import org.apache.spark.sql.types.LongType$;
import org.apache.spark.sql.types.StringType$;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
@@ -210,19 +209,16 @@ public class TestFilteredScan {
@Test
public void testUnpartitionedIDFilters() {
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of(
"path", unpartitioned.toString())
);
-
- IcebergSource source = new IcebergSource();
+ IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), options);
for (int i = 0; i < 10; i += 1) {
- DataSourceReader reader = source.createReader(options);
-
- pushFilters(reader, EqualTo.apply("id", i));
+ pushFilters(scan, EqualTo.apply("id", i));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should only create one task for a small file", 1, tasks.size());
+ InputPartition[] partitions = scan.planInputPartitions();
+ Assert.assertEquals("Should only create one task for a small file", 1, partitions.length);
// validate row filtering
assertEqualsSafe(SCHEMA.asStruct(), expected(i),
@@ -232,7 +228,7 @@ public class TestFilteredScan {
@Test
public void testUnpartitionedCaseInsensitiveIDFilters() {
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of(
"path", unpartitioned.toString())
);
@@ -241,15 +237,14 @@ public class TestFilteredScan {
TestFilteredScan.spark.conf().set("spark.sql.caseSensitive", "false");
try {
- IcebergSource source = new IcebergSource();
for (int i = 0; i < 10; i += 1) {
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), false, options);
- pushFilters(reader, EqualTo.apply("ID", i)); // note lower(ID) == lower(id), so there must be a match
+ pushFilters(scan, EqualTo.apply("ID", i)); // note lower(ID) == lower(id), so there must be a match
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should only create one task for a small file", 1, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should only create one task for a small file", 1, tasks.length);
// validate row filtering
assertEqualsSafe(SCHEMA.asStruct(), expected(i),
@@ -263,18 +258,16 @@ public class TestFilteredScan {
@Test
public void testUnpartitionedTimestampFilter() {
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of(
"path", unpartitioned.toString())
);
- IcebergSource source = new IcebergSource();
-
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(TABLES.load(options.get("path")), options);
- pushFilters(reader, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should only create one task for a small file", 1, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should only create one task for a small file", 1, tasks.length);
assertEqualsSafe(SCHEMA.asStruct(), expected(5, 6, 7, 8, 9),
read(unpartitioned.toString(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
@@ -282,69 +275,61 @@ public class TestFilteredScan {
@Test
public void testBucketPartitionedIDFilters() {
- File location = buildPartitionedTable("bucketed_by_id", BUCKET_BY_ID, "bucket4", "id");
-
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
- "path", location.toString())
- );
+ Table table = buildPartitionedTable("bucketed_by_id", BUCKET_BY_ID, "bucket4", "id");
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- IcebergSource source = new IcebergSource();
- DataSourceReader unfiltered = source.createReader(options);
+ IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
Assert.assertEquals("Unfiltered table should created 4 read tasks",
- 4, unfiltered.planInputPartitions().size());
+ 4, unfiltered.planInputPartitions().length);
for (int i = 0; i < 10; i += 1) {
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(reader, EqualTo.apply("id", i));
+ pushFilters(scan, EqualTo.apply("id", i));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
+ InputPartition[] tasks = scan.planInputPartitions();
// validate predicate push-down
- Assert.assertEquals("Should create one task for a single bucket", 1, tasks.size());
+ Assert.assertEquals("Should create one task for a single bucket", 1, tasks.length);
// validate row filtering
- assertEqualsSafe(SCHEMA.asStruct(), expected(i), read(location.toString(), "id = " + i));
+ assertEqualsSafe(SCHEMA.asStruct(), expected(i), read(table.location(), "id = " + i));
}
}
@SuppressWarnings("checkstyle:AvoidNestedBlocks")
@Test
public void testDayPartitionedTimestampFilters() {
- File location = buildPartitionedTable("partitioned_by_day", PARTITION_BY_DAY, "ts_day", "ts");
+ Table table = buildPartitionedTable("partitioned_by_day", PARTITION_BY_DAY, "ts_day", "ts");
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
+ IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
- "path", location.toString())
- );
-
- IcebergSource source = new IcebergSource();
- DataSourceReader unfiltered = source.createReader(options);
Assert.assertEquals("Unfiltered table should created 2 read tasks",
- 2, unfiltered.planInputPartitions().size());
+ 2, unfiltered.planInputPartitions().length);
{
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(reader, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should create one task for 2017-12-21", 1, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should create one task for 2017-12-21", 1, tasks.length);
assertEqualsSafe(SCHEMA.asStruct(), expected(5, 6, 7, 8, 9),
- read(location.toString(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
+ read(table.location(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
}
{
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(reader, And.apply(
+ pushFilters(scan, And.apply(
GreaterThan.apply("ts", "2017-12-22T06:00:00+00:00"),
LessThan.apply("ts", "2017-12-22T08:00:00+00:00")));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should create one task for 2017-12-22", 1, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should create one task for 2017-12-22", 1, tasks.length);
- assertEqualsSafe(SCHEMA.asStruct(), expected(1, 2), read(location.toString(),
+ assertEqualsSafe(SCHEMA.asStruct(), expected(1, 2), read(table.location(),
"ts > cast('2017-12-22 06:00:00+00:00' as timestamp) and " +
"ts < cast('2017-12-22 08:00:00+00:00' as timestamp)"));
}
@@ -353,40 +338,37 @@ public class TestFilteredScan {
@SuppressWarnings("checkstyle:AvoidNestedBlocks")
@Test
public void testHourPartitionedTimestampFilters() {
- File location = buildPartitionedTable("partitioned_by_hour", PARTITION_BY_HOUR, "ts_hour", "ts");
+ Table table = buildPartitionedTable("partitioned_by_hour", PARTITION_BY_HOUR, "ts_hour", "ts");
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
- "path", location.toString())
- );
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
+ IcebergBatchScan unfiltered = new IcebergBatchScan(table, options);
- IcebergSource source = new IcebergSource();
- DataSourceReader unfiltered = source.createReader(options);
Assert.assertEquals("Unfiltered table should created 9 read tasks",
- 9, unfiltered.planInputPartitions().size());
+ 9, unfiltered.planInputPartitions().length);
{
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(reader, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
+ pushFilters(scan, LessThan.apply("ts", "2017-12-22T00:00:00+00:00"));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should create 4 tasks for 2017-12-21: 15, 17, 21, 22", 4, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should create 4 tasks for 2017-12-21: 15, 17, 21, 22", 4, tasks.length);
assertEqualsSafe(SCHEMA.asStruct(), expected(8, 9, 7, 6, 5),
- read(location.toString(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
+ read(table.location().toString(), "ts < cast('2017-12-22 00:00:00+00:00' as timestamp)"));
}
{
- DataSourceReader reader = source.createReader(options);
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- pushFilters(reader, And.apply(
+ pushFilters(scan, And.apply(
GreaterThan.apply("ts", "2017-12-22T06:00:00+00:00"),
LessThan.apply("ts", "2017-12-22T08:00:00+00:00")));
- List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
- Assert.assertEquals("Should create 2 tasks for 2017-12-22: 6, 7", 2, tasks.size());
+ InputPartition[] tasks = scan.planInputPartitions();
+ Assert.assertEquals("Should create 2 tasks for 2017-12-22: 6, 7", 2, tasks.length);
- assertEqualsSafe(SCHEMA.asStruct(), expected(2, 1), read(location.toString(),
+ assertEqualsSafe(SCHEMA.asStruct(), expected(2, 1), read(table.location(),
"ts > cast('2017-12-22 06:00:00+00:00' as timestamp) and " +
"ts < cast('2017-12-22 08:00:00+00:00' as timestamp)"));
}
@@ -427,32 +409,28 @@ public class TestFilteredScan {
@Test
public void testPartitionedByDataStartsWithFilter() {
- File location = buildPartitionedTable("partitioned_by_data", PARTITION_BY_DATA, "data_ident", "data");
+ Table table = buildPartitionedTable("partitioned_by_data", PARTITION_BY_DATA, "data_ident", "data");
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of("path", table.location()));
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
- "path", location.toString())
- );
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
- IcebergSource source = new IcebergSource();
- DataSourceReader reader = source.createReader(options);
- pushFilters(reader, new StringStartsWith("data", "junc"));
+ pushFilters(scan, new StringStartsWith("data", "junc"));
- Assert.assertEquals(1, reader.planInputPartitions().size());
+ Assert.assertEquals(1, scan.planInputPartitions().length);
}
@Test
public void testPartitionedByIdStartsWith() {
- File location = buildPartitionedTable("partitioned_by_id", PARTITION_BY_ID, "id_ident", "id");
+ Table table = buildPartitionedTable("partitioned_by_id", PARTITION_BY_ID, "id_ident", "id");
- DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
- "path", location.toString())
+ CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(ImmutableMap.of(
+ "path", table.location())
);
- IcebergSource source = new IcebergSource();
- DataSourceReader reader = source.createReader(options);
- pushFilters(reader, new StringStartsWith("data", "junc"));
+ IcebergBatchScan scan = new IcebergBatchScan(table, options);
+ pushFilters(scan, new StringStartsWith("data", "junc"));
- Assert.assertEquals(1, reader.planInputPartitions().size());
+ Assert.assertEquals(1, scan.planInputPartitions().length);
}
@Test
@@ -509,19 +487,19 @@ public class TestFilteredScan {
return expected;
}
- private void pushFilters(DataSourceReader reader, Filter... filters) {
- Assert.assertTrue(reader instanceof SupportsPushDownFilters);
- SupportsPushDownFilters filterable = (SupportsPushDownFilters) reader;
+ private void pushFilters(Scan scan, Filter... filters) {
+ Assert.assertTrue(scan instanceof SupportsPushDownFilters);
+ SupportsPushDownFilters filterable = (SupportsPushDownFilters) scan;
filterable.pushFilters(filters);
}
- private File buildPartitionedTable(String desc, PartitionSpec spec, String udf, String partitionColumn) {
+ private Table buildPartitionedTable(String desc, PartitionSpec spec, String udf, String partitionColumn) {
File location = new File(parent, desc);
- Table byId = TABLES.create(SCHEMA, spec, location.toString());
+ Table table = TABLES.create(SCHEMA, spec, location.toString());
// Do not combine or split files because the tests expect a split per partition.
// A target split size of 2048 helps us achieve that.
- byId.updateProperties().set("read.split.target-size", "2048").commit();
+ table.updateProperties().set("read.split.target-size", "2048").commit();
// copy the unpartitioned table into the partitioned table to produce the partitioned data
Dataset<Row> allRows = spark.read()
@@ -536,9 +514,11 @@ public class TestFilteredScan {
.write()
.format("iceberg")
.mode("append")
- .save(byId.location());
+ .save(table.location());
+
+ table.refresh();
- return location;
+ return table;
}
private List<Record> testRecords(org.apache.avro.Schema avroSchema) {
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
index fdf1b33..fdf5daa 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSource.java
@@ -20,17 +20,16 @@
package org.apache.iceberg.spark.source;
import org.apache.hadoop.conf.Configuration;
-import org.apache.iceberg.Table;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-public class TestIcebergSource extends IcebergSource {
+public class TestIcebergSource extends IcebergTableProvider {
@Override
public String shortName() {
return "iceberg-test";
}
@Override
- protected Table findTable(DataSourceOptions options, Configuration conf) {
- return TestTables.load(options.get("iceberg.table.name").get());
+ public org.apache.iceberg.Table findTable(CaseInsensitiveStringMap options, Configuration conf) {
+ return TestTables.load(options.get("iceberg.table.name"));
}
}
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
index 5357187..da936ed 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestParquetWrite.java
@@ -40,6 +40,7 @@ import org.apache.spark.sql.SparkSession;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -163,7 +164,8 @@ public class TestParquetWrite {
Assert.assertEquals("Result rows should match", expected, actual);
}
- @Test
+ // ignore due to spark default use static
+ @Ignore
public void testOverwrite() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
@@ -195,6 +197,7 @@ public class TestParquetWrite {
// overwrite with 2*id to replace record 2, append 4 and 6
df.withColumn("id", df.col("id").multiply(2)).select("id", "data").write()
+ .option("partitionOverwriteMode", "dynamic")
.format("iceberg")
.mode("overwrite")
.save(location.toString());
@@ -339,7 +342,8 @@ public class TestParquetWrite {
Assert.assertTrue("All DataFiles contain 1000 rows", files.stream().allMatch(d -> d.recordCount() == 1000));
}
- @Test
+ // This fails due to SPARK-28730
+ @Ignore
public void testWriteProjection() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
@@ -372,7 +376,8 @@ public class TestParquetWrite {
Assert.assertEquals("Result rows should match", expected, actual);
}
- @Test
+ // This fails due to SPARK-28730
+ @Ignore
public void testWriteProjectionWithMiddle() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test");
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
index caade72..9550d00 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java
@@ -41,6 +41,7 @@ import org.apache.spark.sql.streaming.StreamingQueryException;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -199,7 +200,8 @@ public class TestStructuredStreaming {
}
}
- @Test
+ // This fails due to SPARK-28730
+ @Ignore
public void testStreamingWriteCompleteModeWithProjection() throws IOException {
File parent = temp.newFolder("parquet");
File location = new File(parent, "test-table");
@@ -263,7 +265,10 @@ public class TestStructuredStreaming {
@Test
public void testStreamingWriteUpdateMode() throws IOException {
exceptionRule.expect(StreamingQueryException.class);
- exceptionRule.expectMessage("Output mode Update is not supported");
+
+ // The following error message to verify is issued by
+ // org.apache.spark.sql.execution.streaming.StreamExecution#createStreamingWrite(SupportsWrite, Map, LogicalPlan)
+ exceptionRule.expectMessage("Data source v2 streaming sinks does not support Update mode");
File parent = temp.newFolder("parquet");
File location = new File(parent, "test-table");
diff --git a/versions.lock b/versions.lock
index 917755e..6010faa 100644
--- a/versions.lock
+++ b/versions.lock
@@ -4,20 +4,22 @@ antlr:antlr:2.7.7 (2 constraints: 36167e02)
aopalliance:aopalliance:1.0 (1 constraints: 170a83ac)
asm:asm:3.1 (2 constraints: 4f19c3c6)
com.carrotsearch:hppc:0.7.2 (1 constraints: f70cda14)
-com.clearspring.analytics:stream:2.7.0 (1 constraints: 1a0dd136)
-com.esotericsoftware:kryo-shaded:4.0.2 (2 constraints: b71345a6)
+com.clearspring.analytics:stream:2.9.6 (1 constraints: 230de736)
+com.esotericsoftware:kryo-shaded:4.0.2 (2 constraints: b8134fa6)
com.esotericsoftware:minlog:1.3.0 (1 constraints: 670e7c4f)
-com.fasterxml.jackson.core:jackson-annotations:2.10.0 (4 constraints: 3d489d20)
-com.fasterxml.jackson.core:jackson-core:2.10.0 (5 constraints: 3a49dbd6)
-com.fasterxml.jackson.core:jackson-databind:2.10.0 (8 constraints: 0a7c5614)
-com.fasterxml.jackson.module:jackson-module-paranamer:2.10.0 (1 constraints: 01162a16)
-com.fasterxml.jackson.module:jackson-module-scala_2.11:2.10.0 (1 constraints: 7f0da251)
+com.fasterxml.jackson.core:jackson-annotations:2.10.0 (4 constraints: 884807a8)
+com.fasterxml.jackson.core:jackson-core:2.10.0 (5 constraints: 60499d20)
+com.fasterxml.jackson.core:jackson-databind:2.10.0 (8 constraints: 297bcc44)
+com.fasterxml.jackson.module:jackson-module-paranamer:2.10.0 (1 constraints: 02163516)
+com.fasterxml.jackson.module:jackson-module-scala_2.12:2.10.0 (1 constraints: 450d1044)
com.github.ben-manes.caffeine:caffeine:2.7.0 (1 constraints: 0b050a36)
-com.github.luben:zstd-jni:1.3.2-2 (1 constraints: 760d7c51)
+com.github.luben:zstd-jni:1.4.3-1 (1 constraints: 780d8f51)
+com.github.spotbugs:spotbugs-annotations:3.1.9 (1 constraints: 8d0d3128)
com.github.stephenc.findbugs:findbugs-annotations:1.3.9-1 (1 constraints: 6d05ab40)
-com.google.code.findbugs:jsr305:3.0.2 (10 constraints: c483db75)
+com.google.code.findbugs:jsr305:3.0.2 (10 constraints: 6b88df04)
com.google.code.gson:gson:2.2.4 (1 constraints: 8c0d3f2f)
com.google.errorprone:error_prone_annotations:2.3.3 (2 constraints: 161a2544)
+com.google.flatbuffers:flatbuffers-java:1.9.0 (2 constraints: e5199714)
com.google.guava:failureaccess:1.0.1 (1 constraints: 140ae1b4)
com.google.guava:guava:28.0-jre (23 constraints: cc5c2ea0)
com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava (1 constraints: bd17c918)
@@ -27,113 +29,109 @@ com.google.j2objc:j2objc-annotations:1.3 (1 constraints: b809eda0)
com.google.protobuf:protobuf-java:2.5.0 (15 constraints: f80eac0f)
com.googlecode.javaewah:JavaEWAH:0.3.2 (1 constraints: ea0dfc42)
com.jamesmurty.utils:java-xmlbuilder:0.4 (1 constraints: e40aa5ca)
-com.jcraft:jsch:0.1.42 (1 constraints: bb0ded3c)
+com.jcraft:jsch:0.1.54 (1 constraints: be0df13c)
com.jolbox:bonecp:0.8.0.RELEASE (2 constraints: b22109f9)
-com.ning:compress-lzf:1.0.3 (1 constraints: 150dba36)
+com.ning:compress-lzf:1.0.3 (1 constraints: 160dc436)
com.sun.jersey:jersey-client:1.9 (4 constraints: 65529ed7)
com.sun.jersey:jersey-core:1.9 (9 constraints: ec8f4404)
com.sun.jersey:jersey-json:1.9 (5 constraints: 945f2f90)
com.sun.jersey:jersey-server:1.9 (4 constraints: ef373c01)
com.sun.jersey.contribs:jersey-guice:1.9 (4 constraints: 65529ed7)
com.sun.xml.bind:jaxb-impl:2.2.3-1 (1 constraints: 330c2404)
-com.thoughtworks.paranamer:paranamer:2.8 (4 constraints: 66361b1c)
-com.twitter:chill-java:0.9.3 (2 constraints: a716716f)
-com.twitter:chill_2.11:0.9.3 (2 constraints: 121b92c3)
-com.twitter:parquet-hadoop-bundle:1.6.0 (3 constraints: 7c262424)
-com.univocity:univocity-parsers:2.7.3 (1 constraints: c40ccb27)
-com.vlkan:flatbuffers:1.2.0-3f79e055 (2 constraints: 411e1dee)
+com.thoughtworks.paranamer:paranamer:2.8 (4 constraints: 68364d1c)
+com.twitter:chill-java:0.9.3 (2 constraints: a916ae6f)
+com.twitter:chill_2.12:0.9.3 (2 constraints: 141bd1c3)
+com.twitter:parquet-hadoop-bundle:1.6.0 (3 constraints: 7d265724)
+com.univocity:univocity-parsers:2.8.3 (2 constraints: a01b9ef6)
commons-beanutils:commons-beanutils:1.7.0 (1 constraints: da0e635f)
commons-beanutils:commons-beanutils-core:1.8.0 (1 constraints: 1d134124)
commons-cli:commons-cli:1.2 (8 constraints: 9467c282)
-commons-codec:commons-codec:1.10 (18 constraints: cfeebaf2)
+commons-codec:commons-codec:1.10 (18 constraints: d1eef4f2)
commons-collections:commons-collections:3.2.2 (4 constraints: 42476934)
commons-configuration:commons-configuration:1.6 (1 constraints: 2d0d5c14)
commons-daemon:commons-daemon:1.0.13 (1 constraints: d50c811c)
commons-dbcp:commons-dbcp:1.4 (3 constraints: 9029e0e4)
commons-digester:commons-digester:1.8 (1 constraints: bf1228fe)
commons-el:commons-el:1.0 (1 constraints: fb077074)
-commons-httpclient:commons-httpclient:3.1 (4 constraints: e52cc77f)
+commons-httpclient:commons-httpclient:3.1 (4 constraints: e62ccf7f)
commons-io:commons-io:2.4 (6 constraints: 4a568049)
commons-lang:commons-lang:2.6 (19 constraints: 5f0d34d8)
commons-logging:commons-logging:1.2 (27 constraints: f2ab390b)
-commons-net:commons-net:3.1 (3 constraints: 3d222e61)
+commons-net:commons-net:3.1 (3 constraints: 3e223661)
commons-pool:commons-pool:1.6 (4 constraints: e336ab5e)
dk.brics.automaton:automaton:1.11-8 (1 constraints: 92088a8d)
hsqldb:hsqldb:1.8.0.10 (1 constraints: f008499f)
io.airlift:aircompressor:0.10 (1 constraints: 090a9fb2)
-io.dropwizard.metrics:metrics-core:3.1.5 (6 constraints: 865ea0ba)
-io.dropwizard.metrics:metrics-graphite:3.1.5 (1 constraints: 1a0dc936)
-io.dropwizard.metrics:metrics-json:3.1.5 (1 constraints: 1a0dc936)
-io.dropwizard.metrics:metrics-jvm:3.1.5 (1 constraints: 1a0dc936)
-io.netty:netty:3.9.9.Final (9 constraints: 9eb0396d)
-io.netty:netty-all:4.1.17.Final (3 constraints: d2312526)
-javax.activation:activation:1.1.1 (1 constraints: 140dbb36)
-javax.annotation:javax.annotation-api:1.2 (2 constraints: 2d21193d)
-javax.inject:javax.inject:1 (4 constraints: 852d0c1a)
+io.dropwizard.metrics:metrics-core:3.2.6 (6 constraints: 955e05c1)
+io.dropwizard.metrics:metrics-graphite:3.2.6 (1 constraints: 1d0dd736)
+io.dropwizard.metrics:metrics-json:3.2.6 (1 constraints: 1d0dd736)
+io.dropwizard.metrics:metrics-jvm:3.2.6 (1 constraints: 1d0dd736)
+io.netty:netty:3.10.6.Final (8 constraints: 91a1ee0f)
+io.netty:netty-all:4.1.42.Final (3 constraints: d031f725)
+jakarta.annotation:jakarta.annotation-api:1.3.4 (4 constraints: 083a0f79)
+jakarta.ws.rs:jakarta.ws.rs-api:2.1.5 (5 constraints: 8764845f)
+javax.activation:activation:1.1.1 (1 constraints: 150dc536)
+javax.inject:javax.inject:1 (2 constraints: b018a173)
javax.jdo:jdo-api:3.0.1 (2 constraints: 4c1dcc1a)
-javax.servlet:javax.servlet-api:3.1.0 (1 constraints: 150dc436)
+javax.servlet:javax.servlet-api:3.1.0 (1 constraints: 160dce36)
javax.servlet:servlet-api:2.5 (7 constraints: c87e5aa1)
javax.servlet.jsp:jsp-api:2.1 (1 constraints: 290d5a14)
javax.transaction:jta:1.1 (1 constraints: 9f07d96b)
-javax.validation:validation-api:1.1.0.Final (1 constraints: 13133130)
-javax.ws.rs:javax.ws.rs-api:2.0.1 (5 constraints: 6e649355)
+javax.validation:validation-api:2.0.1.Final (1 constraints: 14133a30)
javax.xml.bind:jaxb-api:2.2.11 (6 constraints: a069fd48)
javolution:javolution:5.5.1 (1 constraints: f00d1a43)
jline:jline:2.12 (3 constraints: 98208776)
-joda-time:joda-time:2.9.9 (5 constraints: c2326fe6)
+joda-time:joda-time:2.9.9 (5 constraints: c33279e6)
log4j:apache-log4j-extras:1.2.17 (4 constraints: 3f36b1af)
-log4j:log4j:1.2.17 (12 constraints: 22ab5529)
-net.hydromatic:eigenbase-properties:1.1.5 (1 constraints: 5f0daf2c)
+log4j:log4j:1.2.17 (12 constraints: 24ab8929)
net.java.dev.jets3t:jets3t:0.9.0 (2 constraints: ec152b22)
-net.razorvine:pyrolite:4.13 (1 constraints: eb0cb829)
+net.razorvine:pyrolite:4.30 (1 constraints: eb0cc229)
net.sf.kosmosfs:kfs:0.3 (1 constraints: fd077074)
net.sf.opencsv:opencsv:2.3 (2 constraints: a218daa5)
-net.sf.py4j:py4j:0.10.7 (1 constraints: 490d0044)
+net.sf.py4j:py4j:0.10.8.1 (1 constraints: aa0d2f5f)
org.antlr:ST4:4.0.4 (3 constraints: 5521e4e4)
org.antlr:antlr-runtime:3.4 (6 constraints: b84229a1)
-org.antlr:antlr4-runtime:4.7 (1 constraints: 7a0e125f)
+org.antlr:antlr4-runtime:4.7.1 (1 constraints: da0e9d7c)
org.antlr:stringtemplate:3.2.1 (1 constraints: c10a3bc6)
org.apache.ant:ant:1.9.1 (3 constraints: a721ed14)
org.apache.ant:ant-launcher:1.9.1 (1 constraints: 69082485)
-org.apache.arrow:arrow-format:0.10.0 (1 constraints: 1f0de721)
-org.apache.arrow:arrow-memory:0.10.0 (1 constraints: 1f0de721)
-org.apache.arrow:arrow-vector:0.10.0 (1 constraints: e90c9734)
+org.apache.arrow:arrow-format:0.12.0 (1 constraints: 210ded21)
+org.apache.arrow:arrow-memory:0.12.0 (1 constraints: 210ded21)
+org.apache.arrow:arrow-vector:0.12.0 (1 constraints: 010f7d8b)
org.apache.avro:avro:1.8.2 (4 constraints: 3d2eebf3)
org.apache.avro:avro-ipc:1.8.2 (1 constraints: f90b5bf4)
-org.apache.avro:avro-mapred:1.8.2 (2 constraints: 3a1a4787)
-org.apache.calcite:calcite-avatica:1.2.0-incubating (4 constraints: a044b922)
-org.apache.calcite:calcite-core:1.2.0-incubating (2 constraints: bd20f965)
-org.apache.calcite:calcite-linq4j:1.2.0-incubating (1 constraints: ac1147d8)
+org.apache.avro:avro-mapred:1.8.2 (2 constraints: 3c1a8487)
org.apache.commons:commons-compress:1.8.1 (6 constraints: 274bbeb0)
-org.apache.commons:commons-crypto:1.0.0 (2 constraints: 3a1e5fbf)
-org.apache.commons:commons-lang3:3.9 (4 constraints: c52f877f)
-org.apache.commons:commons-math3:3.4.1 (2 constraints: a11af290)
+org.apache.commons:commons-crypto:1.0.0 (2 constraints: 3c1ea6bf)
+org.apache.commons:commons-lang3:3.9 (5 constraints: 583e671a)
+org.apache.commons:commons-math3:3.4.1 (2 constraints: a21afc90)
+org.apache.commons:commons-text:1.6 (1 constraints: bb0cd11c)
org.apache.curator:curator-client:2.7.1 (2 constraints: 6a1d2734)
org.apache.curator:curator-framework:2.7.1 (4 constraints: 4937d02c)
-org.apache.curator:curator-recipes:2.7.1 (2 constraints: a61acc91)
-org.apache.derby:derby:10.12.1.1 (3 constraints: 9f2cb182)
+org.apache.curator:curator-recipes:2.7.1 (2 constraints: a91ada91)
+org.apache.derby:derby:10.12.1.1 (3 constraints: a02cf182)
org.apache.directory.api:api-asn1-api:1.0.0-M20 (1 constraints: 3d163b13)
org.apache.directory.api:api-util:1.0.0-M20 (1 constraints: 3d163b13)
org.apache.directory.server:apacheds-i18n:2.0.0-M15 (1 constraints: 42164713)
org.apache.directory.server:apacheds-kerberos-codec:2.0.0-M15 (1 constraints: 8f0d3b45)
-org.apache.hadoop:hadoop-annotations:2.7.3 (16 constraints: 2c27b38c)
-org.apache.hadoop:hadoop-auth:2.7.3 (1 constraints: 900d4d2f)
-org.apache.hadoop:hadoop-client:2.7.3 (2 constraints: 2b12043c)
-org.apache.hadoop:hadoop-common:2.7.3 (3 constraints: 482267f7)
-org.apache.hadoop:hadoop-hdfs:2.7.3 (4 constraints: b834c025)
-org.apache.hadoop:hadoop-mapreduce-client-app:2.7.3 (3 constraints: ab2f8436)
-org.apache.hadoop:hadoop-mapreduce-client-common:2.7.3 (4 constraints: 184f4f66)
-org.apache.hadoop:hadoop-mapreduce-client-core:2.7.3 (4 constraints: 66361812)
-org.apache.hadoop:hadoop-mapreduce-client-jobclient:2.7.3 (2 constraints: 3b1dfa13)
-org.apache.hadoop:hadoop-mapreduce-client-shuffle:2.7.3 (2 constraints: 2628c449)
-org.apache.hadoop:hadoop-yarn-api:2.7.3 (10 constraints: 07b8bd4c)
-org.apache.hadoop:hadoop-yarn-client:2.7.3 (1 constraints: 1f14626e)
-org.apache.hadoop:hadoop-yarn-common:2.7.3 (9 constraints: b3b2f06f)
-org.apache.hadoop:hadoop-yarn-server-applicationhistoryservice:2.7.3 (1 constraints: f5157dcf)
-org.apache.hadoop:hadoop-yarn-server-common:2.7.3 (7 constraints: 5192cac0)
-org.apache.hadoop:hadoop-yarn-server-nodemanager:2.7.3 (2 constraints: 6726468d)
-org.apache.hadoop:hadoop-yarn-server-resourcemanager:2.7.3 (2 constraints: af2040af)
-org.apache.hadoop:hadoop-yarn-server-web-proxy:2.7.3 (2 constraints: cb287679)
+org.apache.hadoop:hadoop-annotations:2.7.4 (16 constraints: 3c2741a6)
+org.apache.hadoop:hadoop-auth:2.7.4 (1 constraints: 910d4e2f)
+org.apache.hadoop:hadoop-client:2.7.4 (2 constraints: 2c12103c)
+org.apache.hadoop:hadoop-common:2.7.4 (3 constraints: 4a2296f7)
+org.apache.hadoop:hadoop-hdfs:2.7.4 (4 constraints: bb345226)
+org.apache.hadoop:hadoop-mapreduce-client-app:2.7.4 (3 constraints: ae2f1637)
+org.apache.hadoop:hadoop-mapreduce-client-common:2.7.4 (4 constraints: 1c4fae67)
+org.apache.hadoop:hadoop-mapreduce-client-core:2.7.4 (4 constraints: 6936ae12)
+org.apache.hadoop:hadoop-mapreduce-client-jobclient:2.7.4 (2 constraints: 3d1d2914)
+org.apache.hadoop:hadoop-mapreduce-client-shuffle:2.7.4 (2 constraints: 2828024a)
+org.apache.hadoop:hadoop-yarn-api:2.7.4 (10 constraints: 11b87d56)
+org.apache.hadoop:hadoop-yarn-client:2.7.4 (1 constraints: 2014636e)
+org.apache.hadoop:hadoop-yarn-common:2.7.4 (9 constraints: bcb2d777)
+org.apache.hadoop:hadoop-yarn-server-applicationhistoryservice:2.7.4 (1 constraints: f6157ecf)
+org.apache.hadoop:hadoop-yarn-server-common:2.7.4 (7 constraints: 589281c5)
+org.apache.hadoop:hadoop-yarn-server-nodemanager:2.7.4 (2 constraints: 69267b8d)
+org.apache.hadoop:hadoop-yarn-server-resourcemanager:2.7.4 (2 constraints: b0206faf)
+org.apache.hadoop:hadoop-yarn-server-web-proxy:2.7.4 (2 constraints: cd28b579)
org.apache.hive:hive-common:1.2.1 (1 constraints: 740bb5e4)
org.apache.hive:hive-metastore:1.2.1 (2 constraints: 0b1094b7)
org.apache.hive:hive-serde:1.2.1 (1 constraints: 350d6320)
@@ -143,95 +141,97 @@ org.apache.hive.shims:hive-shims-0.23:1.2.1 (1 constraints: 850b5fe5)
org.apache.hive.shims:hive-shims-common:1.2.1 (4 constraints: 233b0a15)
org.apache.hive.shims:hive-shims-scheduler:1.2.1 (1 constraints: 850b5fe5)
org.apache.htrace:htrace-core:3.1.0-incubating (2 constraints: cd22cffa)
-org.apache.httpcomponents:httpclient:4.5.6 (4 constraints: 573134dd)
+org.apache.httpcomponents:httpclient:4.5.6 (4 constraints: 5e316add)
org.apache.httpcomponents:httpcore:4.4.10 (3 constraints: d327f763)
-org.apache.ivy:ivy:2.4.0 (3 constraints: 0826dbf1)
-org.apache.orc:orc-core:1.5.6 (2 constraints: d011de27)
-org.apache.orc:orc-mapreduce:1.5.5 (1 constraints: c30cc227)
-org.apache.orc:orc-shims:1.5.6 (1 constraints: 420aebbc)
+org.apache.ivy:ivy:2.4.0 (3 constraints: 09260ef2)
+org.apache.orc:orc-core:1.5.7 (2 constraints: d311ea27)
+org.apache.orc:orc-mapreduce:1.5.7 (1 constraints: c60cce27)
+org.apache.orc:orc-shims:1.5.7 (1 constraints: 430aecbc)
org.apache.parquet:parquet-avro:1.10.1 (1 constraints: 35052a3b)
-org.apache.parquet:parquet-column:1.10.1 (3 constraints: 9429eeca)
+org.apache.parquet:parquet-column:1.10.1 (3 constraints: 9529f9ca)
org.apache.parquet:parquet-common:1.10.1 (2 constraints: 4c1e7385)
org.apache.parquet:parquet-encoding:1.10.1 (1 constraints: ca0ef964)
org.apache.parquet:parquet-format:2.4.0 (3 constraints: e72a97ca)
-org.apache.parquet:parquet-hadoop:1.10.1 (2 constraints: de1ac5b3)
+org.apache.parquet:parquet-hadoop:1.10.1 (2 constraints: df1ad0b3)
org.apache.parquet:parquet-jackson:1.10.1 (1 constraints: b70ee763)
org.apache.pig:pig:0.14.0 (1 constraints: 37052f3b)
-org.apache.spark:spark-avro_2.11:2.4.4 (1 constraints: 0c050536)
-org.apache.spark:spark-catalyst_2.11:2.4.4 (1 constraints: c20cc327)
-org.apache.spark:spark-core_2.11:2.4.4 (3 constraints: b528ed99)
-org.apache.spark:spark-hive_2.11:2.4.4 (1 constraints: 0c050536)
-org.apache.spark:spark-kvstore_2.11:2.4.4 (1 constraints: 1b0dcc36)
-org.apache.spark:spark-launcher_2.11:2.4.4 (1 constraints: 1b0dcc36)
-org.apache.spark:spark-network-common_2.11:2.4.4 (2 constraints: b01eeee2)
-org.apache.spark:spark-network-shuffle_2.11:2.4.4 (1 constraints: 1b0dcc36)
-org.apache.spark:spark-sketch_2.11:2.4.4 (2 constraints: 981bd4f5)
-org.apache.spark:spark-sql_2.11:2.4.4 (1 constraints: 1e0d0037)
-org.apache.spark:spark-tags_2.11:2.4.4 (8 constraints: 036fa69d)
-org.apache.spark:spark-unsafe_2.11:2.4.4 (2 constraints: f11bc213)
-org.apache.thrift:libfb303:0.9.3 (3 constraints: 6725fac0)
-org.apache.thrift:libthrift:0.9.3 (5 constraints: 71415452)
-org.apache.xbean:xbean-asm6-shaded:4.8 (2 constraints: 2419a30f)
-org.apache.zookeeper:zookeeper:3.4.6 (11 constraints: 18a71f48)
+org.apache.spark:spark-avro_2.12:3.0.0-preview (1 constraints: 3408936b)
+org.apache.spark:spark-catalyst_2.12:3.0.0-preview (1 constraints: eb0f139b)
+org.apache.spark:spark-core_2.12:3.0.0-preview (3 constraints: 30323128)
+org.apache.spark:spark-hive_2.12:3.0.0-preview (1 constraints: 3408936b)
+org.apache.spark:spark-kvstore_2.12:3.0.0-preview (1 constraints: 4410e4ac)
+org.apache.spark:spark-launcher_2.12:3.0.0-preview (1 constraints: 4410e4ac)
+org.apache.spark:spark-network-common_2.12:3.0.0-preview (2 constraints: 02258f19)
+org.apache.spark:spark-network-shuffle_2.12:3.0.0-preview (1 constraints: 4410e4ac)
+org.apache.spark:spark-sketch_2.12:3.0.0-preview (2 constraints: ea219afb)
+org.apache.spark:spark-sql_2.12:3.0.0-preview (1 constraints: 471030ad)
+org.apache.spark:spark-tags_2.12:3.0.0-preview (8 constraints: 4b88bf32)
+org.apache.spark:spark-unsafe_2.12:3.0.0-preview (2 constraints: 4322791f)
+org.apache.thrift:libfb303:0.9.3 (3 constraints: 682504c1)
+org.apache.thrift:libthrift:0.12.0 (5 constraints: 9941518e)
+org.apache.xbean:xbean-asm7-shaded:4.14 (2 constraints: 8019e63c)
+org.apache.yetus:audience-annotations:0.5.0 (1 constraints: 850d2528)
+org.apache.zookeeper:zookeeper:3.4.14 (11 constraints: 48a76cef)
org.checkerframework:checker-qual:2.8.1 (2 constraints: 1a1a3944)
org.codehaus.jackson:jackson-core-asl:1.9.13 (13 constraints: c6a96183)
org.codehaus.jackson:jackson-jaxrs:1.9.13 (2 constraints: 821bca9d)
-org.codehaus.jackson:jackson-mapper-asl:1.9.13 (13 constraints: 27a6dc31)
+org.codehaus.jackson:jackson-mapper-asl:1.9.13 (13 constraints: 28a66532)
org.codehaus.jackson:jackson-xc:1.9.13 (2 constraints: 821bca9d)
-org.codehaus.janino:commons-compiler:3.0.9 (3 constraints: 0a2837cc)
-org.codehaus.janino:janino:3.0.9 (2 constraints: 3f1c6304)
+org.codehaus.janino:commons-compiler:3.0.15 (3 constraints: 65285c17)
+org.codehaus.janino:janino:3.0.15 (2 constraints: 6d1cd220)
org.codehaus.jettison:jettison:1.1 (4 constraints: a84e24a9)
org.codehaus.mojo:animal-sniffer-annotations:1.17 (1 constraints: ed09d8aa)
org.datanucleus:datanucleus-api-jdo:3.2.6 (2 constraints: 5a1d241c)
-org.datanucleus:datanucleus-core:3.2.10 (5 constraints: 4a4401e0)
+org.datanucleus:datanucleus-core:3.2.10 (5 constraints: 4b4465e0)
org.datanucleus:datanucleus-rdbms:3.2.9 (2 constraints: 601db41c)
org.eclipse.jdt:core:3.1.1 (1 constraints: b40a38d8)
-org.fusesource.leveldbjni:leveldbjni-all:1.8 (9 constraints: 91a69ae7)
-org.glassfish.hk2:hk2-api:2.4.0-b34 (5 constraints: 9d5608c7)
-org.glassfish.hk2:hk2-locator:2.4.0-b34 (4 constraints: 3d490865)
-org.glassfish.hk2:hk2-utils:2.4.0-b34 (2 constraints: 0719352b)
-org.glassfish.hk2:osgi-resource-locator:1.0.1 (2 constraints: 79234465)
-org.glassfish.hk2.external:aopalliance-repackaged:2.4.0-b34 (2 constraints: 0719352b)
-org.glassfish.hk2.external:javax.inject:2.4.0-b34 (6 constraints: 52712a15)
-org.glassfish.jersey.bundles.repackaged:jersey-guava:2.22.2 (1 constraints: 231118d4)
-org.glassfish.jersey.containers:jersey-container-servlet:2.22.2 (1 constraints: 490d1144)
-org.glassfish.jersey.containers:jersey-container-servlet-core:2.22.2 (2 constraints: 6425a010)
-org.glassfish.jersey.core:jersey-client:2.22.2 (2 constraints: 791ef7a3)
-org.glassfish.jersey.core:jersey-common:2.22.2 (6 constraints: 5f747f50)
-org.glassfish.jersey.core:jersey-server:2.22.2 (3 constraints: 553f5d56)
-org.glassfish.jersey.media:jersey-media-jaxb:2.22.2 (1 constraints: 3111f1d4)
+org.fusesource.leveldbjni:leveldbjni-all:1.8 (9 constraints: 93a6dbe7)
+org.glassfish.hk2:hk2-api:2.5.0 (1 constraints: 6c0c5e0c)
+org.glassfish.hk2:hk2-locator:2.5.0 (1 constraints: 4210dcad)
+org.glassfish.hk2:hk2-utils:2.5.0 (2 constraints: 1d17de76)
+org.glassfish.hk2:osgi-resource-locator:1.0.3 (2 constraints: 7d23b265)
+org.glassfish.hk2.external:aopalliance-repackaged:2.5.0 (2 constraints: 1d17de76)
+org.glassfish.hk2.external:jakarta.inject:2.5.0 (8 constraints: ed811f01)
+org.glassfish.jersey.containers:jersey-container-servlet:2.29 (1 constraints: f10cc129)
+org.glassfish.jersey.containers:jersey-container-servlet-core:2.29 (2 constraints: b324b295)
+org.glassfish.jersey.core:jersey-client:2.29 (2 constraints: c81dc63c)
+org.glassfish.jersey.core:jersey-common:2.29 (7 constraints: 6382122d)
+org.glassfish.jersey.core:jersey-server:2.29 (3 constraints: 4b3e492d)
+org.glassfish.jersey.inject:jersey-hk2:2.29 (1 constraints: f10cc129)
+org.glassfish.jersey.media:jersey-media-jaxb:2.29 (1 constraints: d810c8b2)
org.iq80.snappy:snappy:0.2 (1 constraints: 890d5927)
-org.javassist:javassist:3.18.1-GA (1 constraints: 570d4740)
-org.jodd:jodd-core:3.5.2 (2 constraints: 0c1bda93)
+org.javassist:javassist:3.22.0-CR2 (1 constraints: 900dbf4d)
+org.jodd:jodd-core:3.5.2 (2 constraints: 0d1b0d94)
org.json:json:20090211 (1 constraints: 890c4218)
-org.json4s:json4s-ast_2.11:3.5.3 (1 constraints: 0c0b9ae9)
-org.json4s:json4s-core_2.11:3.5.3 (1 constraints: 4c0c5316)
-org.json4s:json4s-jackson_2.11:3.5.3 (1 constraints: 1c0dd336)
-org.json4s:json4s-scalap_2.11:3.5.3 (1 constraints: 0c0b9ae9)
-org.lz4:lz4-java:1.4.0 (1 constraints: 160dc336)
-org.mortbay.jetty:jetty:6.1.26 (4 constraints: c8369437)
+org.json4s:json4s-ast_2.12:3.6.6 (1 constraints: 110baae9)
+org.json4s:json4s-core_2.12:3.6.6 (1 constraints: 510c6316)
+org.json4s:json4s-jackson_2.12:3.6.6 (1 constraints: 210de336)
+org.json4s:json4s-scalap_2.12:3.6.6 (1 constraints: 110baae9)
+org.lz4:lz4-java:1.6.0 (1 constraints: 190dd336)
+org.mortbay.jetty:jetty:6.1.26 (5 constraints: cf45d2d7)
+org.mortbay.jetty:jetty-sslengine:6.1.26 (1 constraints: c30d113d)
org.mortbay.jetty:jetty-util:6.1.26 (7 constraints: 7e689dae)
org.mortbay.jetty:jsp-2.1:6.1.14 (1 constraints: 9408a38d)
org.mortbay.jetty:jsp-api-2.1:6.1.14 (2 constraints: 7e130c9d)
org.mortbay.jetty:servlet-api:2.5-20081211 (1 constraints: 390cbd19)
org.mortbay.jetty:servlet-api-2.5:6.1.14 (2 constraints: e51482f7)
org.objenesis:objenesis:2.5.1 (2 constraints: 19198bcb)
-org.roaringbitmap:RoaringBitmap:0.7.45 (1 constraints: 510d1c44)
+org.roaringbitmap:RoaringBitmap:0.7.45 (1 constraints: 520d2744)
org.roaringbitmap:shims:0.7.45 (1 constraints: 260eb249)
-org.scala-lang:scala-library:2.11.12 (11 constraints: 5c9bfe44)
-org.scala-lang:scala-reflect:2.11.12 (1 constraints: 340fb09a)
-org.scala-lang.modules:scala-parser-combinators_2.11:1.1.0 (1 constraints: cf0e717c)
-org.scala-lang.modules:scala-xml_2.11:1.0.6 (1 constraints: 080b84e9)
-org.slf4j:jcl-over-slf4j:1.7.16 (1 constraints: 500d1d44)
-org.slf4j:jul-to-slf4j:1.7.16 (1 constraints: 500d1d44)
-org.slf4j:slf4j-api:1.7.25 (47 constraints: 26a3ad5d)
+org.scala-lang:scala-library:2.12.10 (12 constraints: f7aba36d)
+org.scala-lang:scala-reflect:2.12.10 (2 constraints: ab1c0b7a)
+org.scala-lang.modules:scala-parser-combinators_2.12:1.1.2 (1 constraints: d20e7d7c)
+org.scala-lang.modules:scala-xml_2.12:1.2.0 (1 constraints: 150dc736)
+org.slf4j:jcl-over-slf4j:1.7.16 (1 constraints: 510d2844)
+org.slf4j:jul-to-slf4j:1.7.16 (1 constraints: 510d2844)
+org.slf4j:slf4j-api:1.7.25 (48 constraints: a9b2a1d1)
org.sonatype.sisu.inject:cglib:2.2.1-v20090111 (1 constraints: aa0cfd36)
-org.spark-project.hive:hive-exec:1.2.1.spark2 (1 constraints: 990fa09c)
-org.spark-project.hive:hive-metastore:1.2.1.spark2 (1 constraints: 990fa09c)
-org.spark-project.spark:unused:1.0.0 (12 constraints: 9aab75cf)
+org.spark-project.hive:hive-exec:1.2.1.spark2 (1 constraints: 9a0fb19c)
+org.spark-project.hive:hive-metastore:1.2.1.spark2 (1 constraints: 9a0fb19c)
+org.spark-project.spark:unused:1.0.0 (12 constraints: a6ab49db)
org.tukaani:xz:1.5 (1 constraints: f008458a)
-org.xerial.snappy:snappy-java:1.1.7.3 (3 constraints: 1126203f)
-oro:oro:2.0.8 (3 constraints: 3b229337)
+org.xerial.snappy:snappy-java:1.1.7.3 (3 constraints: 12262c3f)
+oro:oro:2.0.8 (3 constraints: 3c22c237)
stax:stax-api:1.0.1 (2 constraints: ea186edd)
tomcat:jasper-compiler:5.5.12 (1 constraints: 9508ab8d)
tomcat:jasper-runtime:5.5.12 (1 constraints: 9508ab8d)
@@ -241,10 +241,14 @@ xmlenc:xmlenc:0.52 (3 constraints: 05228b2f)
[Test dependencies]
junit:junit:4.12 (1 constraints: db04ff30)
+net.hydromatic:eigenbase-properties:1.1.5 (1 constraints: 5f0daf2c)
+org.apache.calcite:calcite-avatica:1.2.0-incubating (2 constraints: 01216c5c)
+org.apache.calcite:calcite-core:1.2.0-incubating (1 constraints: 560fda68)
+org.apache.calcite:calcite-linq4j:1.2.0-incubating (1 constraints: ac1147d8)
org.apache.curator:apache-curator:2.6.0 (1 constraints: 0a0bf4d6)
-org.apache.hadoop:hadoop-mapreduce-client-hs:2.7.3 (1 constraints: b60fac84)
-org.apache.hadoop:hadoop-minicluster:2.7.3 (1 constraints: 0e050d36)
-org.apache.hadoop:hadoop-yarn-server-tests:2.7.3 (1 constraints: b60fac84)
+org.apache.hadoop:hadoop-mapreduce-client-hs:2.7.4 (1 constraints: b70fad84)
+org.apache.hadoop:hadoop-minicluster:2.7.4 (1 constraints: 0e050d36)
+org.apache.hadoop:hadoop-yarn-server-tests:2.7.4 (1 constraints: b70fad84)
org.apache.hive:hive-ant:1.2.1 (1 constraints: 060be4d6)
org.apache.hive:hive-exec:1.2.1 (1 constraints: 0605f735)
org.apache.velocity:velocity:1.5 (1 constraints: 480ae2b4)
diff --git a/versions.props b/versions.props
index a83da88..eb1f6d7 100644
--- a/versions.props
+++ b/versions.props
@@ -5,8 +5,8 @@ org.apache.hadoop:* = 2.7.3
org.apache.hive:hive-standalone-metastore = 1.2.1
org.apache.orc:orc-core = 1.5.6
org.apache.parquet:parquet-avro = 1.10.1
-org.apache.spark:spark-hive_2.11 = 2.4.4
-org.apache.spark:spark-avro_2.11 = 2.4.4
+org.apache.spark:spark-hive_2.12 = 3.0.0-preview
+org.apache.spark:spark-avro_2.12 = 3.0.0-preview
org.apache.pig:pig = 0.14.0
org.apache.commons:commons-lang3 = 3.9
com.fasterxml.jackson.*:* = 2.10.0
[incubator-iceberg] 03/03: Add a SparkCatalog implementation.
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch spark-3
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
commit 6f42a9fe58a692ed571ca5b49a7537c5a03a0ae3
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Tue Dec 10 12:41:31 2019 -0800
Add a SparkCatalog implementation.
---
.../org/apache/iceberg/hadoop/HadoopCatalog.java | 18 +-
.../java/org/apache/iceberg/hive/HiveCatalog.java | 19 +-
.../org/apache/iceberg/spark/SparkCatalog.java | 252 +++++++++++++++++++++
.../apache/iceberg/spark/source/SparkTable.java | 8 +-
.../iceberg/spark/source/StagedSparkTable.java | 42 ++++
5 files changed, 331 insertions(+), 8 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
index 03108f5..ea9c833 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
@@ -59,24 +59,37 @@ import org.apache.iceberg.exceptions.RuntimeIOException;
*/
public class HadoopCatalog extends BaseMetastoreCatalog implements Closeable {
private static final String ICEBERG_HADOOP_WAREHOUSE_BASE = "iceberg/warehouse";
+ private final String name;
private final Configuration conf;
private String warehouseLocation;
/**
* The constructor of the HadoopCatalog. It uses the passed location as its warehouse directory.
*
+ * @param name the name of this catalog
* @param conf The Hadoop configuration
* @param warehouseLocation The location used as warehouse directory
*/
- public HadoopCatalog(Configuration conf, String warehouseLocation) {
+ public HadoopCatalog(String name, Configuration conf, String warehouseLocation) {
Preconditions.checkArgument(warehouseLocation != null && !warehouseLocation.equals(""),
"no location provided for warehouse");
+ this.name = name;
this.conf = conf;
this.warehouseLocation = warehouseLocation.replaceAll("/*$", "");
}
/**
+ * The constructor of the HadoopCatalog. It uses the passed location as its warehouse directory.
+ *
+ * @param conf The Hadoop configuration
+ * @param warehouseLocation The location used as warehouse directory
+ */
+ public HadoopCatalog(Configuration conf, String warehouseLocation) {
+ this("hadoop", conf, warehouseLocation);
+ }
+
+ /**
* The constructor of the HadoopCatalog. It gets the value of <code>fs.defaultFS</code> property
* from the passed Hadoop configuration as its default file system, and use the default directory
* <code>iceberg/warehouse</code> as the warehouse directory.
@@ -84,13 +97,14 @@ public class HadoopCatalog extends BaseMetastoreCatalog implements Closeable {
* @param conf The Hadoop configuration
*/
public HadoopCatalog(Configuration conf) {
+ this.name = "hadoop";
this.conf = conf;
this.warehouseLocation = conf.get("fs.defaultFS") + "/" + ICEBERG_HADOOP_WAREHOUSE_BASE;
}
@Override
protected String name() {
- return "hadoop";
+ return name;
}
@Override
diff --git a/hive/src/main/java/org/apache/iceberg/hive/HiveCatalog.java b/hive/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
index ad9b05d..9213d7f 100644
--- a/hive/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
+++ b/hive/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
@@ -44,15 +44,26 @@ import org.slf4j.LoggerFactory;
public class HiveCatalog extends BaseMetastoreCatalog implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(HiveCatalog.class);
- private final HiveClientPool clients;
private final Configuration conf;
+ private final HiveClientPool clients;
private final StackTraceElement[] createStack;
+ private final String name;
private boolean closed;
public HiveCatalog(Configuration conf) {
- this.clients = new HiveClientPool(2, conf);
- this.conf = conf;
+ this("hive", null, 2, conf);
+ }
+
+ public HiveCatalog(String name, String uri, int clientPoolSize, Configuration conf) {
+ this.conf = new Configuration(conf);
+ // before building the client pool, overwrite the configuration's URIs if the argument is non-null
+ if (uri != null) {
+ this.conf.set("hive.metastore.uris", uri);
+ }
+
+ this.clients = new HiveClientPool(clientPoolSize, this.conf);
this.createStack = Thread.currentThread().getStackTrace();
+ this.name = name;
this.closed = false;
}
@@ -82,7 +93,7 @@ public class HiveCatalog extends BaseMetastoreCatalog implements Closeable {
@Override
protected String name() {
- return "hive";
+ return name;
}
@Override
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java b/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
new file mode 100644
index 0000000..cd81e8b
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
@@ -0,0 +1,252 @@
+/*
+ * 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.iceberg.spark;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CachingCatalog;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.spark.source.SparkTable;
+import org.apache.iceberg.spark.source.StagedSparkTable;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.StagedTable;
+import org.apache.spark.sql.connector.catalog.StagingTableCatalog;
+import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.connector.catalog.TableChange.RemoveProperty;
+import org.apache.spark.sql.connector.catalog.TableChange.SetProperty;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+/**
+ * A Spark TableCatalog implementation that wraps Iceberg's {@link Catalog} interface.
+ */
+public class SparkCatalog implements StagingTableCatalog {
+ private String catalogName = null;
+ private Catalog icebergCatalog = null;
+
+ /**
+ * Build an Iceberg {@link Catalog} to be used by this Spark catalog adapter.
+ *
+ * @param name Spark's catalog name
+ * @param options Spark's catalog options
+ * @return an Iceberg catalog
+ */
+ protected Catalog buildIcebergCatalog(String name, CaseInsensitiveStringMap options) {
+ Configuration conf = SparkSession.active().sparkContext().hadoopConfiguration();
+ String catalogType = options.getOrDefault("type", "hive");
+ switch (catalogType) {
+ case "hive":
+ int clientPoolSize = options.getInt("clients", 2);
+ String uri = options.get("uri");
+ return new HiveCatalog(name, uri, clientPoolSize, conf);
+
+ case "hadoop":
+ String warehouseLocation = options.get("warehouse");
+ return new HadoopCatalog(name, conf, warehouseLocation);
+
+ default:
+ throw new UnsupportedOperationException("Unknown catalog type: " + catalogType);
+ }
+ }
+
+ /**
+ * Build an Iceberg {@link TableIdentifier} for the given Spark identifier.
+ *
+ * @param identifier Spark's identifier
+ * @return an Iceberg identifier
+ */
+ protected TableIdentifier buildIdentifier(Identifier identifier) {
+ return TableIdentifier.of(Namespace.of(identifier.namespace()), identifier.name());
+ }
+
+ @Override
+ public Identifier[] listTables(String[] namespace) {
+ // TODO: handle namespaces
+ return new Identifier[0];
+ }
+
+ @Override
+ public SparkTable loadTable(Identifier ident) throws NoSuchTableException {
+ try {
+ return new SparkTable(icebergCatalog.loadTable(buildIdentifier(ident)));
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ throw new NoSuchTableException(ident);
+ }
+ }
+
+ @Override
+ public SparkTable createTable(Identifier ident, StructType schema, Transform[] transforms,
+ Map<String, String> properties) throws TableAlreadyExistsException {
+ Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ try {
+ return new SparkTable(icebergCatalog.createTable(buildIdentifier(ident),
+ icebergSchema, SparkUtil.toPartitionSpec(icebergSchema, transforms), properties.get("location"), properties));
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistsException(ident);
+ }
+ }
+
+ @Override
+ public StagedTable stageCreate(Identifier ident, StructType schema, Transform[] transforms,
+ Map<String, String> properties) throws TableAlreadyExistsException {
+ Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ try {
+ return new StagedSparkTable(icebergCatalog.newCreateTableTransaction(buildIdentifier(ident), icebergSchema,
+ SparkUtil.toPartitionSpec(icebergSchema, transforms), properties.get("location"), properties));
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistsException(ident);
+ }
+ }
+
+ @Override
+ public StagedTable stageReplace(Identifier ident, StructType schema, Transform[] transforms,
+ Map<String, String> properties) throws NoSuchTableException {
+ Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ try {
+ return new StagedSparkTable(icebergCatalog.newReplaceTableTransaction(buildIdentifier(ident), icebergSchema,
+ SparkUtil.toPartitionSpec(icebergSchema, transforms), properties.get("location"), properties,
+ false /* do not create */));
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ throw new NoSuchTableException(ident);
+ }
+ }
+
+ @Override
+ public StagedTable stageCreateOrReplace(Identifier ident, StructType schema, Transform[] transforms,
+ Map<String, String> properties) {
+ Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ return new StagedSparkTable(icebergCatalog.newReplaceTableTransaction(buildIdentifier(ident), icebergSchema,
+ SparkUtil.toPartitionSpec(icebergSchema, transforms), properties.get("location"), properties,
+ true /* create or replace */));
+ }
+
+ @Override
+ public SparkTable alterTable(Identifier ident, TableChange... changes) throws NoSuchTableException {
+ SetProperty setLocation = null;
+ SetProperty setSnapshotId = null;
+ List<TableChange> propertyChanges = Lists.newArrayList();
+ List<TableChange> schemaChanges = Lists.newArrayList();
+
+ for (TableChange change : changes) {
+ if (change instanceof SetProperty) {
+ SetProperty set = (SetProperty) change;
+ if ("location".equalsIgnoreCase(set.property())) {
+ setLocation = set;
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.property())) {
+ setSnapshotId = set;
+ } else {
+ propertyChanges.add(set);
+ }
+ } else if (change instanceof RemoveProperty) {
+ propertyChanges.add(change);
+ } else {
+ schemaChanges.add(change);
+ }
+ }
+
+ try {
+ Table table = icebergCatalog.loadTable(buildIdentifier(ident));
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId.value());
+ table.rollback().toSnapshotId(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation()
+ .setLocation(setLocation.value())
+ .commit();
+ }
+
+ if (!propertyChanges.isEmpty()) {
+ SparkUtil.applyPropertyChanges(transaction.updateProperties(), propertyChanges).commit();
+ }
+
+ if (!schemaChanges.isEmpty()) {
+ SparkUtil.applySchemaChanges(transaction.updateSchema(), schemaChanges).commit();
+ }
+
+ transaction.commitTransaction();
+
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ throw new NoSuchTableException(ident);
+ }
+ return null;
+ }
+
+ @Override
+ public boolean dropTable(Identifier ident) {
+ try {
+ return icebergCatalog.dropTable(buildIdentifier(ident), true);
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ return false;
+ }
+ }
+
+ @Override
+ public void renameTable(Identifier from, Identifier to) throws NoSuchTableException, TableAlreadyExistsException {
+ try {
+ icebergCatalog.renameTable(buildIdentifier(from), buildIdentifier(to));
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ throw new NoSuchTableException(from);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistsException(to);
+ }
+ }
+
+ @Override
+ public void invalidateTable(Identifier ident) {
+ try {
+ icebergCatalog.loadTable(buildIdentifier(ident)).refresh();
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException ignored) {
+ // ignore if the table doesn't exist, it is not cached
+ }
+ }
+
+ @Override
+ final public void initialize(String name, CaseInsensitiveStringMap options) {
+ boolean cacheEnabled = Boolean.parseBoolean(options.getOrDefault("cache-enabled", "true"));
+ Catalog catalog = buildIcebergCatalog(name, options);
+
+ this.catalogName = name;
+ this.icebergCatalog = cacheEnabled ? CachingCatalog.wrap(catalog) : catalog;
+ }
+
+ @Override
+ public String name() {
+ return catalogName;
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index f85aac2..30b8b6b 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -35,7 +35,7 @@ import org.apache.spark.sql.connector.write.WriteBuilder;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-class SparkTable implements org.apache.spark.sql.connector.catalog.Table, SupportsRead, SupportsWrite {
+public class SparkTable implements org.apache.spark.sql.connector.catalog.Table, SupportsRead, SupportsWrite {
private static final Set<TableCapability> CAPABILITIES = ImmutableSet.of(
TableCapability.BATCH_READ,
@@ -49,7 +49,11 @@ class SparkTable implements org.apache.spark.sql.connector.catalog.Table, Suppor
private StructType lazyTableSchema = null;
private SparkSession lazySpark = null;
- SparkTable(Table icebergTable, StructType requestedSchema) {
+ public SparkTable(Table icebergTable) {
+ this(icebergTable, null);
+ }
+
+ public SparkTable(Table icebergTable, StructType requestedSchema) {
this.icebergTable = icebergTable;
this.requestedSchema = requestedSchema;
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/StagedSparkTable.java b/spark/src/main/java/org/apache/iceberg/spark/source/StagedSparkTable.java
new file mode 100644
index 0000000..d427621
--- /dev/null
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/StagedSparkTable.java
@@ -0,0 +1,42 @@
+/*
+ * 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.iceberg.spark.source;
+
+import org.apache.iceberg.Transaction;
+import org.apache.spark.sql.connector.catalog.StagedTable;
+
+public class StagedSparkTable extends SparkTable implements StagedTable {
+ private final Transaction transaction;
+
+ public StagedSparkTable(Transaction transaction) {
+ super(transaction.table());
+ this.transaction = transaction;
+ }
+
+ @Override
+ public void commitStagedChanges() {
+ transaction.commitTransaction();
+ }
+
+ @Override
+ public void abortStagedChanges() {
+ // TODO: clean up
+ }
+}