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 2021/11/01 19:47:20 UTC
[iceberg] branch 0.12.x updated (85d9cda -> ded95a5)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a change to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git.
from 85d9cda Core: Validate concurrently added delete files in OvewriteFiles (#3199)
new b149b10 Avro: Fix file import with correct row count (#3273)
new 3291f77 Build: Fix ErrorProne NewHashMapInt warnings (#3260)
new 94103ea Hive: Fix Catalogs.hiveCatalog method for default catalogs (#3338)
new 4ba2157 Spark: Fix ClassCastException when using bucket UDF (#3368)
new eec6ff6 ORC: Fix importing ORC files with float and double columns and test (#3332)
new ded95a5 Spark 3.2: Remove extra parens to fix checkstyle (#3386)
The 6 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:
api/src/main/java/org/apache/iceberg/Metrics.java | 4 +-
build.gradle | 4 +
.../main/java/org/apache/iceberg/avro/Avro.java | 8 ++
.../apache/iceberg/data/TableMigrationUtil.java | 6 +-
.../main/java/org/apache/iceberg/mr/Catalogs.java | 18 ++-
.../org/apache/iceberg/mr/hive/Deserializer.java | 3 +-
.../apache/iceberg/mr/hive/HiveIcebergSerDe.java | 4 +-
.../java/org/apache/iceberg/mr/TestCatalogs.java | 23 +++-
.../java/org/apache/iceberg/orc/OrcMetrics.java | 37 ++++--
.../org/apache/iceberg/spark/IcebergSpark.java | 3 +-
.../apache/iceberg/spark/SparkValueConverter.java | 3 +-
.../iceberg/spark/source/TestIcebergSpark.java | 132 ++++++++++++++++++-
.../spark/extensions/TestAddFilesProcedure.java | 141 +++++++++++++++++++++
13 files changed, 353 insertions(+), 33 deletions(-)
[iceberg] 03/06: Hive: Fix Catalogs.hiveCatalog method for default
catalogs (#3338)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 94103eaabe556f04103ba2f597c5868289a354cc
Author: pvary <pv...@cloudera.com>
AuthorDate: Wed Oct 27 00:21:43 2021 +0200
Hive: Fix Catalogs.hiveCatalog method for default catalogs (#3338)
---
.../main/java/org/apache/iceberg/mr/Catalogs.java | 14 +++++++++----
.../java/org/apache/iceberg/mr/TestCatalogs.java | 23 +++++++++++++++++++++-
2 files changed, 32 insertions(+), 5 deletions(-)
diff --git a/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java b/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
index 2e54cf5..0bf0473 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
@@ -202,7 +202,15 @@ public final class Catalogs {
*/
public static boolean hiveCatalog(Configuration conf, Properties props) {
String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME);
- return CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE.equalsIgnoreCase(getCatalogType(conf, catalogName));
+ String catalogType = getCatalogType(conf, catalogName);
+ if (catalogType != null) {
+ return CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE.equalsIgnoreCase(catalogType);
+ }
+ catalogType = getCatalogType(conf, ICEBERG_DEFAULT_CATALOG_NAME);
+ if (catalogType != null) {
+ return CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE.equalsIgnoreCase(catalogType);
+ }
+ return getCatalogProperties(conf, catalogName, catalogType).get(CatalogProperties.CATALOG_IMPL) == null;
}
@VisibleForTesting
@@ -279,9 +287,7 @@ public final class Catalogs {
}
} else {
String catalogType = conf.get(InputFormatConfig.CATALOG);
- if (catalogType == null) {
- return CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE;
- } else if (catalogType.equals(LOCATION)) {
+ if (catalogType != null && catalogType.equals(LOCATION)) {
return NO_CATALOG_TYPE;
} else {
return catalogType;
diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestCatalogs.java b/mr/src/test/java/org/apache/iceberg/mr/TestCatalogs.java
index 0d5b1d6..0e76ea1 100644
--- a/mr/src/test/java/org/apache/iceberg/mr/TestCatalogs.java
+++ b/mr/src/test/java/org/apache/iceberg/mr/TestCatalogs.java
@@ -197,6 +197,7 @@ public class TestCatalogs {
Optional<Catalog> defaultCatalog = Catalogs.loadCatalog(conf, null);
Assert.assertTrue(defaultCatalog.isPresent());
Assertions.assertThat(defaultCatalog.get()).isInstanceOf(HiveCatalog.class);
+ Assert.assertTrue(Catalogs.hiveCatalog(conf, new Properties()));
}
@Test
@@ -205,6 +206,7 @@ public class TestCatalogs {
Optional<Catalog> hiveCatalog = Catalogs.loadCatalog(conf, null);
Assert.assertTrue(hiveCatalog.isPresent());
Assertions.assertThat(hiveCatalog.get()).isInstanceOf(HiveCatalog.class);
+ Assert.assertTrue(Catalogs.hiveCatalog(conf, new Properties()));
}
@Test
@@ -214,6 +216,7 @@ public class TestCatalogs {
Optional<Catalog> hadoopCatalog = Catalogs.loadCatalog(conf, null);
Assert.assertTrue(hadoopCatalog.isPresent());
Assertions.assertThat(hadoopCatalog.get()).isInstanceOf(HadoopCatalog.class);
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, new Properties()));
}
@Test
@@ -223,12 +226,14 @@ public class TestCatalogs {
Optional<Catalog> customHadoopCatalog = Catalogs.loadCatalog(conf, null);
Assert.assertTrue(customHadoopCatalog.isPresent());
Assertions.assertThat(customHadoopCatalog.get()).isInstanceOf(CustomHadoopCatalog.class);
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, new Properties()));
}
@Test
public void testLegacyLoadCatalogLocation() {
conf.set(InputFormatConfig.CATALOG, Catalogs.LOCATION);
Assert.assertFalse(Catalogs.loadCatalog(conf, null).isPresent());
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, new Properties()));
}
@Test
@@ -241,9 +246,13 @@ public class TestCatalogs {
@Test
public void testLoadCatalogDefault() {
- Optional<Catalog> defaultCatalog = Catalogs.loadCatalog(conf, "barCatalog");
+ String catalogName = "barCatalog";
+ Optional<Catalog> defaultCatalog = Catalogs.loadCatalog(conf, catalogName);
Assert.assertTrue(defaultCatalog.isPresent());
Assertions.assertThat(defaultCatalog.get()).isInstanceOf(HiveCatalog.class);
+ Properties properties = new Properties();
+ properties.put(InputFormatConfig.CATALOG_NAME, catalogName);
+ Assert.assertTrue(Catalogs.hiveCatalog(conf, properties));
}
@Test
@@ -254,6 +263,9 @@ public class TestCatalogs {
Optional<Catalog> hiveCatalog = Catalogs.loadCatalog(conf, catalogName);
Assert.assertTrue(hiveCatalog.isPresent());
Assertions.assertThat(hiveCatalog.get()).isInstanceOf(HiveCatalog.class);
+ Properties properties = new Properties();
+ properties.put(InputFormatConfig.CATALOG_NAME, catalogName);
+ Assert.assertTrue(Catalogs.hiveCatalog(conf, properties));
}
@Test
@@ -267,6 +279,9 @@ public class TestCatalogs {
Assert.assertTrue(hadoopCatalog.isPresent());
Assertions.assertThat(hadoopCatalog.get()).isInstanceOf(HadoopCatalog.class);
Assert.assertEquals("HadoopCatalog{name=barCatalog, location=/tmp/mylocation}", hadoopCatalog.get().toString());
+ Properties properties = new Properties();
+ properties.put(InputFormatConfig.CATALOG_NAME, catalogName);
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, properties));
}
@Test
@@ -279,6 +294,9 @@ public class TestCatalogs {
Assert.assertTrue(hadoopCatalog.isPresent());
Assertions.assertThat(hadoopCatalog.get()).isInstanceOf(HadoopCatalog.class);
Assert.assertEquals("HadoopCatalog{name=barCatalog, location=/tmp/mylocation}", hadoopCatalog.get().toString());
+ Properties properties = new Properties();
+ properties.put(InputFormatConfig.CATALOG_NAME, catalogName);
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, properties));
}
@Test
@@ -291,6 +309,9 @@ public class TestCatalogs {
Optional<Catalog> customHadoopCatalog = Catalogs.loadCatalog(conf, catalogName);
Assert.assertTrue(customHadoopCatalog.isPresent());
Assertions.assertThat(customHadoopCatalog.get()).isInstanceOf(CustomHadoopCatalog.class);
+ Properties properties = new Properties();
+ properties.put(InputFormatConfig.CATALOG_NAME, catalogName);
+ Assert.assertFalse(Catalogs.hiveCatalog(conf, properties));
}
@Test
[iceberg] 06/06: Spark 3.2: Remove extra parens to fix checkstyle
(#3386)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit ded95a5435f18db9b6367e7aa4d1c666c77cfaed
Author: Kyle Bendickson <98...@users.noreply.github.com>
AuthorDate: Sun Oct 31 11:56:56 2021 -0700
Spark 3.2: Remove extra parens to fix checkstyle (#3386)
---
.../src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
index d85b114..f8ebe21 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
@@ -147,7 +147,7 @@ public abstract class TestIcebergSpark {
spark.sql("SELECT iceberg_bucket_binary_16(X'0020001F')").collectAsList();
Assert.assertEquals(1, results.size());
Assert.assertEquals((int) Transforms.bucket(Types.BinaryType.get(), 16)
- .apply(ByteBuffer.wrap((new byte[]{0x00, 0x20, 0x00, 0x1F}))),
+ .apply(ByteBuffer.wrap(new byte[]{0x00, 0x20, 0x00, 0x1F})),
results.get(0).getInt(0));
}
[iceberg] 04/06: Spark: Fix ClassCastException when using bucket
UDF (#3368)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 4ba2157620d1bbc19c968e69ad43849f3706ba16
Author: Chen Zhang <67...@users.noreply.github.com>
AuthorDate: Wed Oct 27 07:46:09 2021 +0800
Spark: Fix ClassCastException when using bucket UDF (#3368)
---
.../org/apache/iceberg/spark/IcebergSpark.java | 3 +-
.../apache/iceberg/spark/SparkValueConverter.java | 3 +-
.../iceberg/spark/source/TestIcebergSpark.java | 132 +++++++++++++++++++--
3 files changed, 129 insertions(+), 9 deletions(-)
diff --git a/spark/src/main/java/org/apache/iceberg/spark/IcebergSpark.java b/spark/src/main/java/org/apache/iceberg/spark/IcebergSpark.java
index ac659f6..862626d 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/IcebergSpark.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/IcebergSpark.java
@@ -34,6 +34,7 @@ public class IcebergSpark {
SparkTypeToType typeConverter = new SparkTypeToType();
Type sourceIcebergType = typeConverter.atomic(sourceType);
Transform<Object, Integer> bucket = Transforms.bucket(sourceIcebergType, numBuckets);
- session.udf().register(funcName, bucket::apply, DataTypes.IntegerType);
+ session.udf().register(funcName,
+ value -> bucket.apply(SparkValueConverter.convert(sourceIcebergType, value)), DataTypes.IntegerType);
}
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java b/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java
index 92c812a..ef453c0 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkValueConverter.java
@@ -79,8 +79,9 @@ public class SparkValueConverter {
return DateTimeUtils.fromJavaTimestamp((Timestamp) object);
case BINARY:
return ByteBuffer.wrap((byte[]) object);
- case BOOLEAN:
case INTEGER:
+ return ((Number) object).intValue();
+ case BOOLEAN:
case LONG:
case FLOAT:
case DOUBLE:
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
index 14785d7..d85b114 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java
@@ -19,13 +19,22 @@
package org.apache.iceberg.spark.source;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.sql.Date;
+import java.sql.Timestamp;
import java.util.List;
import org.apache.iceberg.spark.IcebergSpark;
import org.apache.iceberg.transforms.Transforms;
import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.util.DateTimeUtils;
+import org.apache.spark.sql.types.CharType;
import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.DecimalType;
+import org.apache.spark.sql.types.VarcharType;
+import org.assertj.core.api.Assertions;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -48,23 +57,132 @@ public abstract class TestIcebergSpark {
}
@Test
- public void testRegisterBucketUDF() {
+ public void testRegisterIntegerBucketUDF() {
IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_int_16", DataTypes.IntegerType, 16);
List<Row> results = spark.sql("SELECT iceberg_bucket_int_16(1)").collectAsList();
Assert.assertEquals(1, results.size());
Assert.assertEquals((int) Transforms.bucket(Types.IntegerType.get(), 16).apply(1),
results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterShortBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_short_16", DataTypes.ShortType, 16);
+ List<Row> results = spark.sql("SELECT iceberg_bucket_short_16(1S)").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.IntegerType.get(), 16).apply(1),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterByteBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_byte_16", DataTypes.ByteType, 16);
+ List<Row> results = spark.sql("SELECT iceberg_bucket_byte_16(1Y)").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.IntegerType.get(), 16).apply(1),
+ results.get(0).getInt(0));
+ }
+ @Test
+ public void testRegisterLongBucketUDF() {
IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_long_16", DataTypes.LongType, 16);
- List<Row> results2 = spark.sql("SELECT iceberg_bucket_long_16(1L)").collectAsList();
- Assert.assertEquals(1, results2.size());
+ List<Row> results = spark.sql("SELECT iceberg_bucket_long_16(1L)").collectAsList();
+ Assert.assertEquals(1, results.size());
Assert.assertEquals((int) Transforms.bucket(Types.LongType.get(), 16).apply(1L),
- results2.get(0).getInt(0));
+ results.get(0).getInt(0));
+ }
+ @Test
+ public void testRegisterStringBucketUDF() {
IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_string_16", DataTypes.StringType, 16);
- List<Row> results3 = spark.sql("SELECT iceberg_bucket_string_16('hello')").collectAsList();
- Assert.assertEquals(1, results3.size());
+ List<Row> results = spark.sql("SELECT iceberg_bucket_string_16('hello')").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.StringType.get(), 16).apply("hello"),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterCharBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_char_16", new CharType(5), 16);
+ List<Row> results = spark.sql("SELECT iceberg_bucket_char_16('hello')").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.StringType.get(), 16).apply("hello"),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterVarCharBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_varchar_16", new VarcharType(5), 16);
+ List<Row> results = spark.sql("SELECT iceberg_bucket_varchar_16('hello')").collectAsList();
+ Assert.assertEquals(1, results.size());
Assert.assertEquals((int) Transforms.bucket(Types.StringType.get(), 16).apply("hello"),
- results3.get(0).getInt(0));
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterDateBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_date_16", DataTypes.DateType, 16);
+ List<Row> results = spark.sql("SELECT iceberg_bucket_date_16(DATE '2021-06-30')").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.DateType.get(), 16)
+ .apply(DateTimeUtils.fromJavaDate(Date.valueOf("2021-06-30"))),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterTimestampBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_timestamp_16", DataTypes.TimestampType, 16);
+ List<Row> results =
+ spark.sql("SELECT iceberg_bucket_timestamp_16(TIMESTAMP '2021-06-30 00:00:00.000')").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.TimestampType.withZone(), 16)
+ .apply(DateTimeUtils.fromJavaTimestamp(Timestamp.valueOf("2021-06-30 00:00:00.000"))),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterBinaryBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_binary_16", DataTypes.BinaryType, 16);
+ List<Row> results =
+ spark.sql("SELECT iceberg_bucket_binary_16(X'0020001F')").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.BinaryType.get(), 16)
+ .apply(ByteBuffer.wrap((new byte[]{0x00, 0x20, 0x00, 0x1F}))),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterDecimalBucketUDF() {
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_decimal_16", new DecimalType(4, 2), 16);
+ List<Row> results =
+ spark.sql("SELECT iceberg_bucket_decimal_16(11.11)").collectAsList();
+ Assert.assertEquals(1, results.size());
+ Assert.assertEquals((int) Transforms.bucket(Types.DecimalType.of(4, 2), 16)
+ .apply(new BigDecimal("11.11")),
+ results.get(0).getInt(0));
+ }
+
+ @Test
+ public void testRegisterBooleanBucketUDF() {
+ Assertions.assertThatThrownBy(() ->
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_boolean_16", DataTypes.BooleanType, 16))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot bucket by type: boolean");
+ }
+
+ @Test
+ public void testRegisterDoubleBucketUDF() {
+ Assertions.assertThatThrownBy(() ->
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_double_16", DataTypes.DoubleType, 16))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot bucket by type: double");
+ }
+
+ @Test
+ public void testRegisterFloatBucketUDF() {
+ Assertions.assertThatThrownBy(() ->
+ IcebergSpark.registerBucketUDF(spark, "iceberg_bucket_float_16", DataTypes.FloatType, 16))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot bucket by type: float");
}
}
[iceberg] 02/06: Build: Fix ErrorProne NewHashMapInt warnings
(#3260)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 3291f771130c876d0559e3bd9c901acc7a8fbcbd
Author: Kyle Bendickson <98...@users.noreply.github.com>
AuthorDate: Sun Oct 10 10:53:21 2021 -0700
Build: Fix ErrorProne NewHashMapInt warnings (#3260)
This should improve performance by allocating the correct size directly rather than reallocating later.
---
api/src/main/java/org/apache/iceberg/Metrics.java | 4 ++--
mr/src/main/java/org/apache/iceberg/mr/Catalogs.java | 4 ++--
mr/src/main/java/org/apache/iceberg/mr/hive/Deserializer.java | 3 ++-
mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java | 4 ++--
4 files changed, 8 insertions(+), 7 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/Metrics.java b/api/src/main/java/org/apache/iceberg/Metrics.java
index d5367c4..8a213ad 100644
--- a/api/src/main/java/org/apache/iceberg/Metrics.java
+++ b/api/src/main/java/org/apache/iceberg/Metrics.java
@@ -24,8 +24,8 @@ import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.nio.ByteBuffer;
-import java.util.HashMap;
import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.util.ByteBuffers;
/**
@@ -230,7 +230,7 @@ public class Metrics implements Serializable {
return null;
} else {
- Map<Integer, ByteBuffer> result = new HashMap<>(size);
+ Map<Integer, ByteBuffer> result = Maps.newHashMapWithExpectedSize(size);
for (int i = 0; i < size; ++i) {
Integer key = (Integer) in.readObject();
diff --git a/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java b/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
index b5cc63b..2e54cf5 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/Catalogs.java
@@ -19,7 +19,6 @@
package org.apache.iceberg.mr;
-import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
@@ -39,6 +38,7 @@ import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
/**
@@ -150,7 +150,7 @@ public final class Catalogs {
String catalogName = props.getProperty(InputFormatConfig.CATALOG_NAME);
// Create a table property map without the controlling properties
- Map<String, String> map = new HashMap<>(props.size());
+ Map<String, String> map = Maps.newHashMapWithExpectedSize(props.size());
for (Object key : props.keySet()) {
if (!PROPERTIES_TO_REMOVE.contains(key)) {
map.put(key.toString(), props.get(key).toString());
diff --git a/mr/src/main/java/org/apache/iceberg/mr/hive/Deserializer.java b/mr/src/main/java/org/apache/iceberg/mr/hive/Deserializer.java
index 458affd..47e9f3e 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/hive/Deserializer.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/hive/Deserializer.java
@@ -33,6 +33,7 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.mr.hive.serde.objectinspector.WriteObjectInspector;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
import org.apache.iceberg.types.Type.PrimitiveType;
import org.apache.iceberg.types.Types.ListType;
@@ -232,7 +233,7 @@ class Deserializer {
FixNameMappingObjectInspectorPair(Schema schema, ObjectInspectorPair pair) {
super(pair.writerInspector(), pair.sourceInspector());
- this.sourceNameMap = new HashMap<>(schema.columns().size());
+ this.sourceNameMap = Maps.newHashMapWithExpectedSize(schema.columns().size());
List<? extends StructField> fields = ((StructObjectInspector) sourceInspector()).getAllStructFieldRefs();
for (int i = 0; i < schema.columns().size(); ++i) {
diff --git a/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java
index 707db6f..ce425c6 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java
@@ -22,7 +22,6 @@ package org.apache.iceberg.mr.hive;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -46,6 +45,7 @@ import org.apache.iceberg.mr.Catalogs;
import org.apache.iceberg.mr.InputFormatConfig;
import org.apache.iceberg.mr.hive.serde.objectinspector.IcebergObjectInspector;
import org.apache.iceberg.mr.mapred.Container;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,7 +55,7 @@ public class HiveIcebergSerDe extends AbstractSerDe {
private ObjectInspector inspector;
private Schema tableSchema;
- private Map<ObjectInspector, Deserializer> deserializers = new HashMap<>(1);
+ private Map<ObjectInspector, Deserializer> deserializers = Maps.newHashMapWithExpectedSize(1);
private Container<Record> row = new Container<>();
@Override
[iceberg] 01/06: Avro: Fix file import with correct row count
(#3273)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit b149b10496bef42f5182768ca3e07edac521b149
Author: Szehon Ho <sz...@gmail.com>
AuthorDate: Wed Oct 20 08:09:10 2021 -0700
Avro: Fix file import with correct row count (#3273)
---
build.gradle | 2 +
.../main/java/org/apache/iceberg/avro/Avro.java | 8 +++
.../apache/iceberg/data/TableMigrationUtil.java | 6 ++-
.../spark/extensions/TestAddFilesProcedure.java | 63 ++++++++++++++++++++++
4 files changed, 78 insertions(+), 1 deletion(-)
diff --git a/build.gradle b/build.gradle
index 8a8812a..e9f5e4b 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1104,6 +1104,8 @@ project(":iceberg-spark3-extensions") {
testCompile project(path: ':iceberg-spark', configuration: 'testArtifacts')
testCompile project(path: ':iceberg-spark3', configuration: 'testArtifacts')
+ testCompile "org.apache.avro:avro"
+
spark31Implementation("org.apache.spark:spark-hive_2.12:${project.ext.Spark31Version}") {
exclude group: 'org.apache.avro', module: 'avro'
exclude group: 'org.apache.arrow'
diff --git a/core/src/main/java/org/apache/iceberg/avro/Avro.java b/core/src/main/java/org/apache/iceberg/avro/Avro.java
index 59715b0..3b4ba48 100644
--- a/core/src/main/java/org/apache/iceberg/avro/Avro.java
+++ b/core/src/main/java/org/apache/iceberg/avro/Avro.java
@@ -636,4 +636,12 @@ public class Avro {
}
}
+ /**
+ * Returns number of rows in specified Avro file
+ * @param file Avro file
+ * @return number of rows in file
+ */
+ public static long rowCount(InputFile file) {
+ return AvroIO.findStartingRowPos(file::newStream, Long.MAX_VALUE);
+ }
}
diff --git a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java
index 2750847..a432c76 100644
--- a/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java
+++ b/data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java
@@ -36,7 +36,9 @@ import org.apache.iceberg.Metrics;
import org.apache.iceberg.MetricsConfig;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.avro.Avro;
import org.apache.iceberg.hadoop.HadoopInputFile;
+import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.orc.OrcMetrics;
import org.apache.iceberg.parquet.ParquetUtil;
@@ -91,7 +93,9 @@ public class TableMigrationUtil {
return Arrays.stream(fs.listStatus(partition, HIDDEN_PATH_FILTER))
.filter(FileStatus::isFile)
.map(stat -> {
- Metrics metrics = new Metrics(-1L, null, null, null);
+ InputFile file = HadoopInputFile.fromLocation(stat.getPath().toString(), conf);
+ long rowCount = Avro.rowCount(file);
+ Metrics metrics = new Metrics(rowCount, null, null, null);
String partitionKey = spec.fields().stream()
.map(PartitionField::name)
.map(name -> String.format("%s=%s", name, partitionPath.get(name)))
diff --git a/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java b/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
index dea01a1..539fb62 100644
--- a/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
+++ b/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
@@ -24,8 +24,17 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.DatumWriter;
import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.DataFile;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
@@ -35,6 +44,7 @@ import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.After;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
@@ -106,6 +116,59 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
sql("SELECT * FROM %s ORDER BY id", tableName));
}
+ @Test
+ public void addAvroFile() throws Exception {
+ // Spark Session Catalog cannot load metadata tables
+ // with "The namespace in session catalog must have exactly one name part"
+ Assume.assumeFalse(catalogName.equals("spark_catalog"));
+
+ // Create an Avro file
+
+ Schema schema = SchemaBuilder.record("record").fields()
+ .requiredInt("id")
+ .requiredString("data")
+ .endRecord();
+ GenericRecord record1 = new GenericData.Record(schema);
+ record1.put("id", 1L);
+ record1.put("data", "a");
+ GenericRecord record2 = new GenericData.Record(schema);
+ record2.put("id", 2L);
+ record2.put("data", "b");
+ File outputFile = temp.newFile("test.avro");
+
+ DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter(schema);
+ DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter(datumWriter);
+ dataFileWriter.create(schema, outputFile);
+ dataFileWriter.append(record1);
+ dataFileWriter.append(record2);
+ dataFileWriter.close();
+
+ String createIceberg =
+ "CREATE TABLE %s (id Long, data String) USING iceberg";
+ sql(createIceberg, tableName);
+
+ Object result = scalarSql("CALL %s.system.add_files('%s', '`avro`.`%s`')",
+ catalogName, tableName, outputFile.getPath());
+ Assert.assertEquals(1L, result);
+
+ List<Object[]> expected = Lists.newArrayList(
+ new Object[]{1L, "a"},
+ new Object[]{2L, "b"}
+ );
+
+ assertEquals("Iceberg table contains correct data",
+ expected,
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+
+ List<Object[]> actualRecordCount = sql("select %s from %s.files",
+ DataFile.RECORD_COUNT.name(),
+ tableName);
+ List<Object[]> expectedRecordCount = Lists.newArrayList();
+ expectedRecordCount.add(new Object[]{2L});
+ assertEquals("Iceberg file metadata should have correct metadata count",
+ expectedRecordCount, actualRecordCount);
+ }
+
// TODO Adding spark-avro doesn't work in tests
@Ignore
public void addDataUnpartitionedAvro() {
[iceberg] 05/06: ORC: Fix importing ORC files with float and double
columns and test (#3332)
Posted by bl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch 0.12.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit eec6ff6e1d1ca956e5520ae11c5ab4f133d468e2
Author: Kyle Bendickson <98...@users.noreply.github.com>
AuthorDate: Fri Oct 29 12:02:24 2021 -0700
ORC: Fix importing ORC files with float and double columns and test (#3332)
---
build.gradle | 2 +
.../java/org/apache/iceberg/orc/OrcMetrics.java | 37 +++++++---
.../spark/extensions/TestAddFilesProcedure.java | 78 ++++++++++++++++++++++
3 files changed, 106 insertions(+), 11 deletions(-)
diff --git a/build.gradle b/build.gradle
index e9f5e4b..40521e7 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1099,6 +1099,8 @@ project(":iceberg-spark3-extensions") {
exclude group: 'org.apache.arrow'
}
+ testCompile project(path: ':iceberg-data', configuration: 'testArtifacts')
+ testCompile project(path: ':iceberg-orc', configuration: 'testArtifacts')
testCompile project(path: ':iceberg-api', configuration: 'testArtifacts')
testCompile project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
testCompile project(path: ':iceberg-spark', configuration: 'testArtifacts')
diff --git a/orc/src/main/java/org/apache/iceberg/orc/OrcMetrics.java b/orc/src/main/java/org/apache/iceberg/orc/OrcMetrics.java
index 0fad4da..a873be5 100644
--- a/orc/src/main/java/org/apache/iceberg/orc/OrcMetrics.java
+++ b/orc/src/main/java/org/apache/iceberg/orc/OrcMetrics.java
@@ -44,7 +44,6 @@ import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.hadoop.HadoopInputFile;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.mapping.NameMapping;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Conversions;
@@ -196,11 +195,17 @@ public class OrcMetrics {
min = Math.toIntExact((long) min);
}
} else if (columnStats instanceof DoubleColumnStatistics) {
- // since Orc includes NaN for upper/lower bounds of floating point columns, and we don't want this behavior,
- // we have tracked metrics for such columns ourselves and thus do not need to rely on Orc's column statistics.
- Preconditions.checkNotNull(fieldMetrics,
- "[BUG] Float or double type columns should have metrics being tracked by Iceberg Orc writers");
- min = fieldMetrics.lowerBound();
+ if (fieldMetrics != null) {
+ // since Orc includes NaN for upper/lower bounds of floating point columns, and we don't want this behavior,
+ // we have tracked metrics for such columns ourselves and thus do not need to rely on Orc's column statistics.
+ min = fieldMetrics.lowerBound();
+ } else {
+ // imported files will not have metrics that were tracked by Iceberg, so fall back to the file's metrics.
+ min = replaceNaN(((DoubleColumnStatistics) columnStats).getMinimum(), Double.NEGATIVE_INFINITY);
+ if (type.typeId() == Type.TypeID.FLOAT) {
+ min = ((Double) min).floatValue();
+ }
+ }
} else if (columnStats instanceof StringColumnStatistics) {
min = ((StringColumnStatistics) columnStats).getMinimum();
} else if (columnStats instanceof DecimalColumnStatistics) {
@@ -234,11 +239,17 @@ public class OrcMetrics {
max = Math.toIntExact((long) max);
}
} else if (columnStats instanceof DoubleColumnStatistics) {
- // since Orc includes NaN for upper/lower bounds of floating point columns, and we don't want this behavior,
- // we have tracked metrics for such columns ourselves and thus do not need to rely on Orc's column statistics.
- Preconditions.checkNotNull(fieldMetrics,
- "[BUG] Float or double type columns should have metrics being tracked by Iceberg Orc writers");
- max = fieldMetrics.upperBound();
+ if (fieldMetrics != null) {
+ // since Orc includes NaN for upper/lower bounds of floating point columns, and we don't want this behavior,
+ // we have tracked metrics for such columns ourselves and thus do not need to rely on Orc's column statistics.
+ max = fieldMetrics.upperBound();
+ } else {
+ // imported files will not have metrics that were tracked by Iceberg, so fall back to the file's metrics.
+ max = replaceNaN(((DoubleColumnStatistics) columnStats).getMaximum(), Double.POSITIVE_INFINITY);
+ if (type.typeId() == Type.TypeID.FLOAT) {
+ max = ((Double) max).floatValue();
+ }
+ }
} else if (columnStats instanceof StringColumnStatistics) {
max = ((StringColumnStatistics) columnStats).getMaximum();
} else if (columnStats instanceof DecimalColumnStatistics) {
@@ -262,6 +273,10 @@ public class OrcMetrics {
return Optional.ofNullable(Conversions.toByteBuffer(type, truncateIfNeeded(Bound.UPPER, type, max, metricsMode)));
}
+ private static Object replaceNaN(double value, double replacement) {
+ return Double.isNaN(value) ? replacement : value;
+ }
+
private static Object truncateIfNeeded(Bound bound, Type type, Object value, MetricsMode metricsMode) {
// Out of the two types which could be truncated, string or binary, ORC only supports string bounds.
// Therefore, truncation will be applied if needed only on string type.
diff --git a/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java b/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
index 539fb62..8096688 100644
--- a/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
+++ b/spark3-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
@@ -33,8 +33,16 @@ import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;
import org.apache.iceberg.AssertHelpers;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
@@ -51,6 +59,8 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+
public class TestAddFilesProcedure extends SparkExtensionsTestBase {
private final String sourceTableName = "source_table";
@@ -507,6 +517,42 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
catalogName, tableName, fileTableDir.getAbsolutePath()));
}
+ @Test
+ public void addOrcFileWithDoubleAndFloatColumns() throws Exception {
+ // Spark Session Catalog cannot load metadata tables
+ // with "The namespace in session catalog must have exactly one name part"
+ Assume.assumeFalse(catalogName.equals("spark_catalog"));
+
+ // Create an ORC file
+ File outputFile = temp.newFile("test.orc");
+ final int numRows = 5;
+ List<Record> expectedRecords = createOrcFile(outputFile, numRows);
+ String createIceberg =
+ "CREATE TABLE %s (x float, y double, z long) USING iceberg";
+ sql(createIceberg, tableName);
+
+ Object result = scalarSql("CALL %s.system.add_files('%s', '`orc`.`%s`')",
+ catalogName, tableName, outputFile.getPath());
+ Assert.assertEquals(1L, result);
+
+ List<Object[]> expected = expectedRecords.stream()
+ .map(record -> new Object[]{record.get(0), record.get(1), record.get(2)})
+ .collect(Collectors.toList());
+
+ // x goes 2.00, 1.99, 1.98, ...
+ assertEquals("Iceberg table contains correct data",
+ expected,
+ sql("SELECT * FROM %s ORDER BY x DESC", tableName));
+
+ List<Object[]> actualRecordCount = sql("select %s from %s.files",
+ DataFile.RECORD_COUNT.name(),
+ tableName);
+ List<Object[]> expectedRecordCount = Lists.newArrayList();
+ expectedRecordCount.add(new Object[]{(long) numRows});
+ assertEquals("Iceberg file metadata should have correct metadata count",
+ expectedRecordCount, actualRecordCount);
+ }
+
private static final StructField[] struct = {
new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
new StructField("name", DataTypes.StringType, false, Metadata.empty()),
@@ -597,4 +643,36 @@ public class TestAddFilesProcedure extends SparkExtensionsTestBase {
partitionedDF.write().insertInto(sourceTableName);
partitionedDF.write().insertInto(sourceTableName);
}
+
+ // Update this to not write a file for import using Iceberg's ID numbers
+ public List<Record> createOrcFile(File orcFile, int numRows) throws IOException {
+ // Needs to be deleted but depend on the rule to delete the file for us again at the end.
+ if (orcFile.exists()) {
+ orcFile.delete();
+ }
+ final org.apache.iceberg.Schema icebergSchema = new org.apache.iceberg.Schema(
+ optional(1, "x", Types.FloatType.get()),
+ optional(2, "y", Types.DoubleType.get()),
+ optional(3, "z", Types.LongType.get())
+ );
+
+ List<Record> records = Lists.newArrayListWithExpectedSize(numRows);
+ for (int i = 0; i < numRows; i += 1) {
+ Record record = org.apache.iceberg.data.GenericRecord.create(icebergSchema);
+ record.setField("x", ((float) (100 - i)) / 100F + 1.0F); // 2.0f, 1.99f, 1.98f, ...
+ record.setField("y", ((double) i) / 100.0D + 2.0D); // 2.0d, 2.01d, 2.02d, ...
+ record.setField("z", 5_000_000_000L + i);
+ records.add(record);
+ }
+
+ OutputFile outFile = Files.localOutput(orcFile);
+ try (FileAppender<Record> appender = org.apache.iceberg.orc.ORC.write(outFile)
+ .schema(icebergSchema)
+ .metricsConfig(MetricsConfig.fromProperties(ImmutableMap.of("write.metadata.metrics.default", "none")))
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .build()) {
+ appender.addAll(records);
+ }
+ return records;
+ }
}