huan233usc commented on code in PR #16711:
URL: https://github.com/apache/iceberg/pull/16711#discussion_r3390439991
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java:
##########
@@ -205,6 +226,23 @@ public StagedTable stageCreate(
Identifier ident, StructType schema, Transform[] transforms, Map<String,
String> properties)
throws TableAlreadyExistsException {
Schema icebergSchema = SparkSchemaUtil.convert(schema);
+ return stageCreate(ident, icebergSchema, transforms, properties);
Review Comment:
Could we add tests for the staged paths — REPLACE TABLE t (id INT, data
STRING DEFAULT 'x') and CREATE OR REPLACE TABLE with a DEFAULT clause?
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkDefaultValues.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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 java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Type;
+import org.apache.spark.sql.connector.catalog.ColumnDefaultValue;
+import org.apache.spark.sql.connector.catalog.DefaultValue;
+import org.apache.spark.sql.connector.expressions.Expression;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
+
+class SparkDefaultValues {
+
+ private SparkDefaultValues() {}
+
+ static Literal<?> toIcebergLiteral(String columnName, Type type,
DefaultValue defaultValue) {
+ if (defaultValue == null) {
+ return null;
+ }
+
+ Expression expression = defaultValue.getExpression();
+ if (expression instanceof
org.apache.spark.sql.connector.expressions.Literal) {
+ org.apache.spark.sql.connector.expressions.Literal<?> literal =
+ (org.apache.spark.sql.connector.expressions.Literal<?>) expression;
+ return toIcebergLiteral(columnName, type, literal.value(),
defaultValue.getSql());
+ } else if (defaultValue instanceof ColumnDefaultValue) {
+ ColumnDefaultValue columnDefault = (ColumnDefaultValue) defaultValue;
+ org.apache.spark.sql.connector.expressions.Literal<?> literal =
columnDefault.getValue();
+ if (literal != null) {
+ return toIcebergLiteral(columnName, type, literal.value(),
defaultValue.getSql());
+ }
+ }
+
+ throw new UnsupportedOperationException(
+ String.format(
+ "Column default for %s must be a literal value: %s",
+ columnName, defaultValue.getSql()));
+ }
+
+ private static Literal<?> toIcebergLiteral(
+ String columnName, Type type, Object sparkValue, String defaultSql) {
+ if (sparkValue == null) {
+ return null;
+ }
+
+ Preconditions.checkArgument(
+ type.isPrimitiveType() && type.typeId() != Type.TypeID.UNKNOWN,
+ "Column default for %s with type %s is not supported by Iceberg Spark
DDL: %s",
+ columnName,
+ type,
+ defaultSql);
+
+ Literal<?> literal = createLiteral(type, sparkValue).to(type);
+ Preconditions.checkArgument(
+ literal != null,
+ "Column default for %s cannot be converted to Iceberg type %s: %s",
+ columnName,
+ type,
+ defaultSql);
+ return literal;
+ }
+
+ private static Literal<?> createLiteral(Type type, Object sparkValue) {
+ if (sparkValue instanceof UTF8String) {
+ return Literal.of(sparkValue.toString());
+ }
+ if (sparkValue instanceof Decimal) {
+ return Literal.of(((Decimal) sparkValue).toJavaBigDecimal());
+ }
+ if (sparkValue instanceof byte[]) {
+ return Literal.of(ByteBuffer.wrap((byte[]) sparkValue));
+ }
+
+ switch (type.typeId()) {
+ case DATE:
+ if (sparkValue instanceof Number) {
+ return Literal.of(((Number) sparkValue).intValue());
+ }
+ break;
+ case TIME:
Review Comment:
Maybe just reject TIME defaults for now, since Spark-side time support is
still WIP in [#16665](https://github.com/apache/iceberg/pull/16665)?
Aside we may also think about the unit here -- Spark 4.1 stores TimeType as
nanoseconds-of-day (https://issues.apache.org/jira/browse/SPARK-52460) while
Iceberg time is microseconds, iirc and this pass-through does no conversion —
so once #16665 makes this path reachable, TIME defaults would be silently
stored 1000x too large.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
