Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Uniform: Fix microsecond conversion for timestamp/ntz partition value #4195

Merged
merged 1 commit into from
Feb 27, 2025
Merged

Uniform: Fix microsecond conversion for timestamp/ntz partition value #4195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...erg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergTransactionUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import shadedForDelta.org.apache.iceberg.{DataFile, DataFiles, FileFormat, Parti
import shadedForDelta.org.apache.iceberg.Metrics
import shadedForDelta.org.apache.iceberg.StructLike
import shadedForDelta.org.apache.iceberg.TableProperties
import shadedForDelta.org.apache.iceberg.util.DateTimeUtil

// scalastyle:off import.ordering.noEmptyLine
import shadedForDelta.org.apache.iceberg.catalog.{Namespace, TableIdentifier => IcebergTableIdentifier}
Expand Down Expand Up @@ -239,7 +240,8 @@ object IcebergTransactionUtils
case _: DecimalType => new java.math.BigDecimal(partitionVal)
case _: BinaryType => ByteBuffer.wrap(partitionVal.getBytes("UTF-8"))
case _: TimestampNTZType =>
java.sql.Timestamp.valueOf(partitionVal).getNanos/1000.asInstanceOf[Long]
DateTimeUtil.isoTimestampToMicros(
partitionVal.replace(" ", "T"))
case _: TimestampType =>
try {
getMicrosSinceEpoch(partitionVal)
Expand All @@ -257,7 +259,8 @@ object IcebergTransactionUtils
}

private def getMicrosSinceEpoch(instant: String): Long = {
Instant.parse(instant).getNano/1000.asInstanceOf[Long]
DateTimeUtil.microsFromInstant(
Instant.parse(instant))
}

private def getMetricsForIcebergDataFile(
Expand Down
6 changes: 3 additions & 3 deletions spark/src/test/scala/org/apache/spark/sql/delta/uniform/UniFormE2ESuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ abstract class UniFormE2EIcebergSuiteBase extends UniFormE2ETest {
"to_date('2016-12-31', 'yyyy-MM-dd')",
"'asdf'",
true,
"TIMESTAMP_NTZ'2021-12-06 00:00:00'",
"TIMESTAMP_NTZ'2021-12-06 05:12:34'",
"TIMESTAMP'2023-08-18 05:00:00UTC-7'"
)

Expand All @@ -88,7 +88,7 @@ abstract class UniFormE2EIcebergSuiteBase extends UniFormE2ETest {
s"where ${partitionColumnName}=${partitionColumnsAndValues._2}"
// Verify against Delta read and Iceberg read
checkAnswer(spark.sql(verificationQuery), Seq(Row(123)))
checkAnswer(createReaderSparkSession.sql(verificationQuery), Seq(Row(123)))
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's important to go through the read function, for some reason when I go through createReaderSparkSession directly that doesn't actually use the Iceberg spark session even though we are overriding it correctly.

assert(read(verificationQuery).sameElements(Seq(Row(123))))
}
}
}
Expand All @@ -109,7 +109,7 @@ abstract class UniFormE2EIcebergSuiteBase extends UniFormE2ETest {
s"where id=1 and ts=TIMESTAMP'2023-08-18 05:00:00UTC-7'"
// Verify against Delta read and Iceberg read
checkAnswer(spark.sql(verificationQuery), Seq(Row(123)))
checkAnswer(createReaderSparkSession.sql(verificationQuery), Seq(Row(123)))
assert(read(verificationQuery).sameElements(Seq(Row(123))))
}
}

Expand Down
Loading