Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions src/DataTypes/Utils.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,12 @@ bool canBeSafelyCast(const DataTypePtr & from_type, const DataTypePtr & to_type)
to_type_unwrapped->getSizeOfValueInMemory() >= from_type->getSizeOfValueInMemory())
return true;

/// An unsigned value fits into a signed integer only if the latter is strictly wider:
/// `UInt8 -> Int16` is lossless, `UInt8 -> Int8` is not.
if (to_which_type.isInt() &&
to_type_unwrapped->getSizeOfValueInMemory() > from_type->getSizeOfValueInMemory())
return true;

if (to_which_type.isString())
return true;

Expand Down
78 changes: 78 additions & 0 deletions src/DataTypes/tests/gtest_can_be_safely_cast.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
#include <DataTypes/DataTypeFactory.h>
#include <DataTypes/Utils.h>

#include <gtest/gtest.h>

using namespace DB;

namespace
{

bool canCast(const std::string & from, const std::string & to)
{
auto & factory = DataTypeFactory::instance();
return canBeSafelyCast(factory.get(from), factory.get(to));
}

}

TEST(CanBeSafelyCast, UnsignedToWiderSigned)
{
EXPECT_TRUE(canCast("UInt8", "Int16"));
EXPECT_TRUE(canCast("UInt8", "Int32"));
EXPECT_TRUE(canCast("UInt8", "Int64"));
EXPECT_TRUE(canCast("UInt8", "Int128"));
EXPECT_TRUE(canCast("UInt8", "Int256"));
EXPECT_TRUE(canCast("UInt16", "Int32"));
EXPECT_TRUE(canCast("UInt16", "Int64"));
EXPECT_TRUE(canCast("UInt32", "Int64"));
EXPECT_TRUE(canCast("UInt64", "Int128"));
EXPECT_TRUE(canCast("UInt64", "Int256"));
EXPECT_TRUE(canCast("UInt128", "Int256"));
}

TEST(CanBeSafelyCast, UnsignedToSignedOfSameOrSmallerSizeIsLossy)
{
EXPECT_FALSE(canCast("UInt8", "Int8"));
EXPECT_FALSE(canCast("UInt16", "Int8"));
EXPECT_FALSE(canCast("UInt16", "Int16"));
EXPECT_FALSE(canCast("UInt32", "Int16"));
EXPECT_FALSE(canCast("UInt32", "Int32"));
EXPECT_FALSE(canCast("UInt64", "Int32"));
EXPECT_FALSE(canCast("UInt64", "Int64"));
EXPECT_FALSE(canCast("UInt128", "Int64"));
EXPECT_FALSE(canCast("UInt128", "Int128"));
EXPECT_FALSE(canCast("UInt256", "Int128"));
EXPECT_FALSE(canCast("UInt256", "Int256"));
}

TEST(CanBeSafelyCast, SignedToUnsignedIsLossy)
{
EXPECT_FALSE(canCast("Int8", "UInt8"));
EXPECT_FALSE(canCast("Int8", "UInt16"));
EXPECT_FALSE(canCast("Int8", "UInt64"));
EXPECT_FALSE(canCast("Int32", "UInt64"));
EXPECT_FALSE(canCast("Int128", "UInt256"));
}

TEST(CanBeSafelyCast, SameSignednessWidening)
{
EXPECT_TRUE(canCast("UInt8", "UInt8"));
EXPECT_TRUE(canCast("UInt8", "UInt64"));
EXPECT_FALSE(canCast("UInt64", "UInt8"));
EXPECT_TRUE(canCast("Int8", "Int64"));
EXPECT_FALSE(canCast("Int64", "Int8"));
}

TEST(CanBeSafelyCast, UnsignedToSignedThroughWrappers)
{
EXPECT_TRUE(canCast("UInt8", "Nullable(Int16)"));
EXPECT_TRUE(canCast("UInt8", "LowCardinality(Int16)"));
EXPECT_TRUE(canCast("Nullable(UInt8)", "Nullable(Int16)"));
EXPECT_TRUE(canCast("Array(UInt8)", "Array(Int16)"));
EXPECT_TRUE(canCast("Tuple(UInt8, UInt32)", "Tuple(Int16, Int64)"));
EXPECT_TRUE(canCast("Map(UInt8, UInt16)", "Map(Int16, Int32)"));

EXPECT_FALSE(canCast("Array(UInt8)", "Array(Int8)"));
EXPECT_FALSE(canCast("Tuple(UInt8, UInt32)", "Tuple(Int16, Int32)"));
}
91 changes: 91 additions & 0 deletions tests/integration/test_export_merge_tree_part_to_iceberg/test.py
Original file line number Diff line number Diff line change
Expand Up @@ -1232,6 +1232,97 @@ def test_export_part_with_castable_widening(cluster):
node.query(f"DROP TABLE IF EXISTS {iceberg}")


def test_export_part_with_unsigned_to_wider_signed_widening(cluster):
"""An unsigned column exported into a strictly wider signed column (UInt8 -> Int16,
UInt16 -> Int32, UInt32 -> Int64, Nullable(UInt8) -> Nullable(Int16)) is a lossless
cast, so it is accepted without export_merge_tree_part_allow_lossy_cast and the
maximum values of the source types round-trip unchanged."""
node = cluster.instances["node1"]
sfx = unique_suffix()
mt = f"mt_unsigned_widen_{sfx}"
iceberg = f"iceberg_unsigned_widen_{sfx}"

make_mt(node, mt, "u8 UInt8, u16 UInt16, u32 UInt32, opt Nullable(UInt8), year Int32", "year")
make_iceberg_s3(node, iceberg, "u8 Int16, u16 Int32, u32 Int64, opt Nullable(Int16), year Int32", "year")

node.query(f"INSERT INTO {mt} VALUES (255, 65535, 4294967295, 255, 2020), (0, 0, 0, NULL, 2020)")
part_2020 = get_part(node, mt, "2020")

export_part(node, mt, part_2020, iceberg)
wait_for_export_part(node, mt, part_2020)

result = node.query(f"SELECT u8, u16, u32, opt, year FROM {iceberg} ORDER BY u8").strip()
assert result == "0\t0\t0\t\\N\t2020\n255\t65535\t4294967295\t255\t2020", (
f"Unexpected data after unsigned -> wider signed export:\n{result}"
)

assert_part_log(node, mt, part_2020)

node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
node.query(f"DROP TABLE IF EXISTS {iceberg}")


def test_export_part_unsigned_partition_column_widened_to_signed(cluster):
"""A partition column that is unsigned in the source (year UInt16) and a strictly wider
signed integer in the destination (year Int32) passes the lossy-cast gate without the
opt-in, and the partition value is preserved."""
node = cluster.instances["node1"]
sfx = unique_suffix()
mt = f"mt_unsigned_pcol_{sfx}"
iceberg = f"iceberg_unsigned_pcol_{sfx}"

make_mt(node, mt, "id Int32, year UInt16", "year")
make_iceberg_s3(node, iceberg, "id Int32, year Int32", "year")

node.query(f"INSERT INTO {mt} VALUES (1, 2020), (2, 2020)")
part_2020 = get_part(node, mt, "2020")

export_part(node, mt, part_2020, iceberg)
wait_for_export_part(node, mt, part_2020)

result = node.query(f"SELECT id, year FROM {iceberg} ORDER BY id").strip()
assert result == "1\t2020\n2\t2020", f"Unexpected data after unsigned partition column export:\n{result}"

assert_part_log(node, mt, part_2020)

node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
node.query(f"DROP TABLE IF EXISTS {iceberg}")


@pytest.mark.parametrize(
"source_type,destination_type",
[("UInt16", "Int16"), ("UInt32", "Int32"), ("UInt64", "Int64")],
)
def test_export_part_unsigned_to_same_size_signed_rejected_without_optin(cluster, source_type, destination_type):
"""An unsigned column exported into a signed column of the same size is lossy (values above
the signed maximum wrap), so it is rejected synchronously with INCOMPATIBLE_COLUMNS unless
export_merge_tree_part_allow_lossy_cast is set."""
node = cluster.instances["node1"]
sfx = unique_suffix()
mt = f"mt_unsigned_same_size_{sfx}"
iceberg = f"iceberg_unsigned_same_size_{sfx}"

make_mt(node, mt, f"id {source_type}, year Int32", "year")
make_iceberg_s3(node, iceberg, f"id {destination_type}, year Int32", "year")

node.query(f"INSERT INTO {mt} VALUES (1, 2020)")
part_2020 = get_part(node, mt, "2020")

error = node.query_and_get_error(
f"ALTER TABLE {mt} EXPORT PART '{part_2020}' TO TABLE {iceberg} "
f"SETTINGS allow_experimental_export_merge_tree_part = 1, "
f"allow_experimental_insert_into_iceberg = 1"
)
assert "INCOMPATIBLE_COLUMNS" in error, f"Expected INCOMPATIBLE_COLUMNS, got: {error!r}"
assert "lossy cast" in error, f"Expected 'lossy cast' in error, got: {error!r}"

count = int(node.query(f"SELECT count() FROM {iceberg}").strip())
assert count == 0, f"Expected no rows after a rejected export, got {count}"

node.query(f"DROP TABLE IF EXISTS {mt} SYNC")
node.query(f"DROP TABLE IF EXISTS {iceberg}")


def test_export_part_with_castable_narrowing_values_fit(cluster):
"""A lossy narrowing (id Int64 -> Int32) succeeds once the user opts in via
export_merge_tree_part_allow_lossy_cast."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -628,6 +628,98 @@ def test_export_partition_with_castable_widening(cluster, source_engine):
)


def test_export_partition_with_unsigned_to_wider_signed_widening(cluster, source_engine):
"""An unsigned column exported into a strictly wider signed column (UInt8 -> Int16,
UInt16 -> Int32, UInt32 -> Int64, Nullable(UInt8) -> Nullable(Int16)) is a lossless
cast, so it is accepted without export_merge_tree_part_allow_lossy_cast and the
maximum values of the source types round-trip unchanged."""
node = cluster.instances["replica1"]

uid = unique_suffix()
mt_table = f"mt_unsigned_widen_{uid}"
iceberg_table = f"iceberg_unsigned_widen_{uid}"

make_source(
node, mt_table, "u8 UInt8, u16 UInt16, u32 UInt32, opt Nullable(UInt8), year Int32", "year",
replica_name="replica1", engine=source_engine,
)
node.query(f"INSERT INTO {mt_table} VALUES (255, 65535, 4294967295, 255, 2020), (0, 0, 0, NULL, 2020)")

make_iceberg_s3(
node, iceberg_table, "u8 Int16, u16 Int32, u32 Int64, opt Nullable(Int16), year Int32", partition_by="year"
)

node.query(
f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {iceberg_table}",
settings={"allow_insert_into_iceberg": 1},
)
wait_for_export_status(node, mt_table, iceberg_table, "2020", "COMPLETED")

result = node.query(f"SELECT u8, u16, u32, opt, year FROM {iceberg_table} ORDER BY u8").strip()
assert result == "0\t0\t0\t\\N\t2020\n255\t65535\t4294967295\t255\t2020", (
f"Unexpected data after unsigned -> wider signed export:\n{result}"
)


def test_export_partition_unsigned_partition_column_widened_to_signed(cluster, source_engine):
"""A partition column that is unsigned in the source (year UInt16) and a strictly wider
signed integer in the destination (year Int32) passes the lossy-cast gate without the
opt-in, and the partition value is preserved."""
node = cluster.instances["replica1"]

uid = unique_suffix()
mt_table = f"mt_unsigned_pcol_{uid}"
iceberg_table = f"iceberg_unsigned_pcol_{uid}"

make_source(node, mt_table, "id Int32, year UInt16", "year", replica_name="replica1", engine=source_engine)
node.query(f"INSERT INTO {mt_table} VALUES (1, 2020), (2, 2020)")

make_iceberg_s3(node, iceberg_table, "id Int32, year Int32", partition_by="year")

node.query(
f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {iceberg_table}",
settings={"allow_insert_into_iceberg": 1},
)
wait_for_export_status(node, mt_table, iceberg_table, "2020", "COMPLETED")

result = node.query(f"SELECT id, year FROM {iceberg_table} ORDER BY id").strip()
assert result == "1\t2020\n2\t2020", f"Unexpected data after unsigned partition column export:\n{result}"


@pytest.mark.parametrize(
"source_type,destination_type",
[("UInt16", "Int16"), ("UInt32", "Int32"), ("UInt64", "Int64")],
)
def test_export_partition_unsigned_to_same_size_signed_rejected_without_optin(
cluster, source_engine, source_type, destination_type
):
"""An unsigned column exported into a signed column of the same size is lossy (values above
the signed maximum wrap), so it is rejected synchronously with INCOMPATIBLE_COLUMNS unless
export_merge_tree_part_allow_lossy_cast is set."""
node = cluster.instances["replica1"]

uid = unique_suffix()
mt_table = f"mt_unsigned_same_size_{uid}"
iceberg_table = f"iceberg_unsigned_same_size_{uid}"

make_source(
node, mt_table, f"id {source_type}, year Int32", "year", replica_name="replica1", engine=source_engine
)
node.query(f"INSERT INTO {mt_table} VALUES (1, 2020)")

make_iceberg_s3(node, iceberg_table, f"id {destination_type}, year Int32", partition_by="year")

error = node.query_and_get_error(
f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {iceberg_table} "
f"SETTINGS allow_insert_into_iceberg = 1"
)
assert "INCOMPATIBLE_COLUMNS" in error, f"Expected INCOMPATIBLE_COLUMNS, got: {error!r}"
assert "lossy cast" in error, f"Expected 'lossy cast' in error, got: {error!r}"

count = int(node.query(f"SELECT count() FROM {iceberg_table}").strip())
assert count == 0, f"Expected no rows after a rejected export, got {count}"


def test_export_partition_with_castable_narrowing_values_fit(cluster, source_engine):
"""A lossy narrowing (id Int64 -> Int32) succeeds once the user opts in via
export_merge_tree_part_allow_lossy_cast."""
Expand Down
Loading