diff --git a/products/warehouse_sources/backend/models/table.py b/products/warehouse_sources/backend/models/table.py index 38d7f0051c52..03ba2679fda5 100644 --- a/products/warehouse_sources/backend/models/table.py +++ b/products/warehouse_sources/backend/models/table.py @@ -107,6 +107,15 @@ class DataWarehouseTableIntrospectedColumn(TypedDict): # subprocess lets us kill it and degrade to the ClickHouse-cluster fallback. CHDB_QUERY_TIMEOUT_SECONDS = 30.0 +# ClickHouse's Hive-style partition inference guesses a type per partition-folder value it +# samples (e.g. our internal `_ph_partition_key`), independently of the physical column type. +# A table whose partition granularity changed over time (see repartition.py's tiering from +# week -> hour) mixes value shapes across folders — e.g. an hour-tier "2017-06-30T05" next to +# older week-tier folders — and CH can misclassify the column as Date, then fail to parse it. +# HogQLGlobalSettings.use_hive_partitioning disables this for the normal HogQL query path; the +# raw ClickHouse queries below bypass that path and must opt out the same way. +DISABLE_HIVE_PARTITIONING_SETTINGS: dict[str, int] = {"use_hive_partitioning": 0} + _CHDB_SUBPROCESS_SCRIPT = """ import sys @@ -441,7 +450,7 @@ def get_columns( quoted_placeholders = {k: escape_param_clickhouse(v) for k, v in placeholder_context.values.items()} # chdb doesn't support parameterized queries - chdb_query = f"DESCRIBE TABLE {s3_table_func}" % quoted_placeholders + chdb_query = f"SET use_hive_partitioning = 0; DESCRIBE TABLE {s3_table_func}" % quoted_placeholders # TODO: upgrade chdb once https://github.com/chdb-io/chdb/issues/342 is actually resolved # See https://github.com/chdb-io/chdb/pull/374 for the fix @@ -472,7 +481,7 @@ def get_columns( attempts = 5 for i in range(attempts): try: - get_columns_settings: dict[str, int] = {} + get_columns_settings: dict[str, int] = dict(DISABLE_HIVE_PARTITIONING_SETTINGS) if self._is_csv_format() and self.csv_allow_double_quotes is not None: get_columns_settings["format_csv_allow_double_quotes"] = ( 1 if self.csv_allow_double_quotes else 0 @@ -531,6 +540,7 @@ def get_max_value_for_column(self, column: str) -> Any | None: result = sync_execute( f"SELECT max({escape_clickhouse_identifier(column)}) FROM {s3_table_func}", args=placeholder_context.values, + settings=DISABLE_HIVE_PARTITIONING_SETTINGS, ) return result[0][0] @@ -564,7 +574,7 @@ def get_count(self, safe_expose_ch_error=True) -> int: quoted_placeholders = {k: escape_param_clickhouse(v) for k, v in placeholder_context.values.items()} # chdb doesn't support parameterized queries - chdb_query = f"SELECT count() FROM {s3_table_func}" % quoted_placeholders + chdb_query = f"SET use_hive_partitioning = 0; SELECT count() FROM {s3_table_func}" % quoted_placeholders chdb_result = run_chdb_query(chdb_query) reader = csv.reader(StringIO(chdb_result)) @@ -585,6 +595,7 @@ def get_count(self, safe_expose_ch_error=True) -> int: result = sync_execute( f"SELECT count() FROM {s3_table_func}", args=placeholder_context.values, + settings=DISABLE_HIVE_PARTITIONING_SETTINGS, ) except Exception as err: capture_exception(err) @@ -880,7 +891,7 @@ def _validate_csv_double_quotes_setting(self) -> None: sync_execute( f"SELECT 1 FROM {func} LIMIT 100", args=ctx.values, - settings={"format_csv_allow_double_quotes": 1 if setting else 0}, + settings={**DISABLE_HIVE_PARTITIONING_SETTINGS, "format_csv_allow_double_quotes": 1 if setting else 0}, ) except ClickHouseServerException as e: if e.code in self._CSV_PARSE_ERROR_CODES: diff --git a/products/warehouse_sources/backend/tests/test_table.py b/products/warehouse_sources/backend/tests/test_table.py index 77e5675cad66..9d1e00a51728 100644 --- a/products/warehouse_sources/backend/tests/test_table.py +++ b/products/warehouse_sources/backend/tests/test_table.py @@ -89,6 +89,35 @@ def test_set_columns_records_order(self) -> None: assert table.column_order == ["z", "a"] +class TestWarehouseQueryDisablesHivePartitioning(BaseTest): + # ClickHouse infers a type for each Hive-style partition-folder value it samples (e.g. our + # internal `_ph_partition_key`) independently of the column's declared type. A table whose + # partition granularity changed over time mixes value shapes across folders (e.g. an + # hour-tier "2017-06-30T05" alongside older week-tier folders), and CH can misclassify the + # column as Date and then fail to parse it — HogQLGlobalSettings disables this inference for + # the normal HogQL query path, so these raw `sync_execute` calls must opt out the same way. + def _table(self) -> DataWarehouseTable: + return DataWarehouseTable(name="t", format="Delta", team=self.team, url_pattern="s3://bucket/team_1/t") + + def test_get_count_disables_hive_partitioning(self) -> None: + with patch( + "products.warehouse_sources.backend.models.table.sync_execute", return_value=[(5,)] + ) as mock_sync_execute: + count = self._table().get_count() + + assert count == 5 + assert mock_sync_execute.call_args.kwargs["settings"]["use_hive_partitioning"] == 0 + + def test_get_max_value_for_column_disables_hive_partitioning(self) -> None: + with patch( + "products.warehouse_sources.backend.models.table.sync_execute", return_value=[(42,)] + ) as mock_sync_execute: + value = self._table().get_max_value_for_column("created_at") + + assert value == 42 + assert mock_sync_execute.call_args.kwargs["settings"]["use_hive_partitioning"] == 0 + + class TestSafeExposeChError: # ClickHouseAtCapacity is a DRF APIException with no `.message`, so the capacity check # must run before the message-matching loop — reordering them would reintroduce an