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

[WIP] [Kernel] New tests for Timestamp_NTZ writes #1

Open
wants to merge 1 commit into
base: tfIntegration
Choose a base branch
from
Open

[WIP] [Kernel] New tests for Timestamp_NTZ writes #1

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
65 changes: 64 additions & 1 deletion kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import io.delta.kernel.expressions.Literal._
import io.delta.kernel.hook.PostCommitHook.PostCommitHookType
import io.delta.kernel.internal.{SnapshotImpl, TableConfig}
import io.delta.kernel.internal.checkpoints.CheckpointerSuite.selectSingleElement
import io.delta.kernel.internal.util.ColumnMapping
import io.delta.kernel.internal.util.{ColumnMapping, VectorUtils}
import io.delta.kernel.internal.util.SchemaUtils.casePreservingPartitionColNames
import io.delta.kernel.types._
import io.delta.kernel.types.DateType.DATE
Expand All @@ -45,6 +45,8 @@ import io.delta.kernel.types.TimestampType.TIMESTAMP
import io.delta.kernel.utils.CloseableIterable
import io.delta.kernel.utils.CloseableIterable.{emptyIterable, inMemoryIterable}

import org.apache.spark.sql.delta.DeltaTableFeatureException

/** Transaction commit in this suite IS REQUIRED TO use commitTransaction than .commit */
class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBase {

Expand Down Expand Up @@ -1323,4 +1325,65 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa
assert(meta.get(ColumnMapping.COLUMN_MAPPING_PHYSICAL_NAME_KEY) == expPhyName)
}
}

test("create table TIMESTAMP_NTZ type should auto-enable the reader/writer feature") {
Copy link

Choose a reason for hiding this comment

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

can we move these tests to DeltaTableFeaturesSuite?

Copy link

Choose a reason for hiding this comment

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

I think there are utilities to check if the protocol contains certain features.

withTempDirAndEngine { (tablePath, engine) =>
val table = Table.forPath(engine, tablePath)
val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)

val testSchema = new StructType().add("tz", TimestampNTZType.TIMESTAMP_NTZ)
val txn = txnBuilder
.withSchema(engine, testSchema)
.build(engine)
val txnResult = commitTransaction(txn, engine, emptyIterable())

assert(txnResult.getVersion === 0)
val protocol = getProtocolActionFromCommit(engine, table, 0)
assert(protocol.isDefined)
assert(protocol.get.getInt(0) == 3)
assert(protocol.get.getInt(1) == 7)
assert(VectorUtils.toJavaList(protocol.get.getArray(3)).contains("timestampNtz"))
}
}

test("create table without TIMESTAMP_NTZ type should use minimum protocol") {
Copy link

Choose a reason for hiding this comment

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

we can combine this test and above one with
Seq(
// Test format: isTimestamp enabled, expected protocol
(true, new Protocol(3, 7, Set("timestampNtz"), Set("timetampNtz"),
(false, new Protocol(1, 2, Set.empty, Set.empty)
) {
case (isTimestampEnabled, expProtocol) => {
... test code ...
}
}

withTempDirAndEngine { (tablePath, engine) =>
val table = Table.forPath(engine, tablePath)
val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)

val txn = txnBuilder
.withSchema(engine, testSchema)
.build(engine)

assert(txn.getSchema(engine) === testSchema)
assert(txn.getPartitionColumns(engine) === Seq.empty.asJava)
val txnResult = commitTransaction(txn, engine, emptyIterable())

assert(txnResult.getVersion === 0)
val protocol = getProtocolActionFromCommit(engine, table, 0)
assert(protocol.isDefined)
assert(protocol.get.getInt(0) == 1)
assert(protocol.get.getInt(1) == 2)
}
}

test("schema evolution from Spark to add TIMESTAMP_NTZ type on a table created with kernel") {
withTempDirAndEngine { (tablePath, engine) =>
val table = Table.forPath(engine, tablePath)
val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)
val txn = txnBuilder
.withSchema(engine, testSchema)
.build(engine)
val txnResult = commitTransaction(txn, engine, emptyIterable())

assert(txnResult.getVersion === 0)
assertThrows[DeltaTableFeatureException] {
spark.sql("ALTER TABLE delta.`" + tablePath + "` ADD COLUMN newCol TIMESTAMP_NTZ")
}
spark.sql("ALTER TABLE delta.`" + tablePath +
"` SET TBLPROPERTIES ('delta.feature.timestampNtz' = 'supported')")
spark.sql("ALTER TABLE delta.`" + tablePath + "` ADD COLUMN newCol TIMESTAMP_NTZ")
}

}
}
Loading