nsivabalan commented on code in PR #20462: URL: https://github.com/apache/hudi/pull/20462#discussion_r4452736418
########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -346,6 +488,182 @@ to complete. Set the same property in `spark.extraConf` for other deployments us this driver. It controls batch size, not total result size or memory; oversized individual rows can still exceed the receive limit. +### SQL, catalog and storage diagnostics + +By default, Kubernetes readiness checks only the TCP listener. To opt into +SQL-aware readiness, first provision a persistent non-empty storage-backed probe +table with a non-null ordinary data column and use values like: + +```yaml +readiness: + sql: + enabled: true + schema: probe_database + table: probe_table + column: probe_column +``` + +The driver image must include `check-ready.py`; rebuild older images. Startup +continues to use TCP so the driver can initialize before the SQL probe runs. +Readiness checks SQL execution, catalog discovery and an actual table-data query +in sequence. A missing or empty probe table makes the pod unready; provision it +before enabling the option. Checks run every 30 seconds with a 30-second timeout. +In the local kind executor test, the probe passed against a pre-populated Hudi +probe table; changing the configured column to a nonexistent name made the pod +unready. Caching and catalog configuration can affect which backing services a +successful read exercises; this check does not prove write access. + +For a separate read-only data-path diagnostic using ADBC, run: + +```bash +python hudi-lakehouse/local-dev/example/connect_readiness.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --schema probe_database --table probe_table --column probe_column +``` + +The command reports SQL, catalog and table-data checks separately and exits +nonzero at the first failure. It requires read access and never modifies the probe +table. A missing/empty table is a failure; caching and catalog configuration can +affect which backing services a successful read actually exercises. This external +diagnostic does not itself change Kubernetes readiness or prove write access. + +### Spark UI, logs and metrics + +For local diagnosis, forward the driver UI to loopback only: + +```bash +kubectl -n hudi-lakehouse port-forward --address 127.0.0.1 deployment/hudi-spark-connect 4040:4040 +curl http://127.0.0.1:4040/api/v1/applications +curl http://127.0.0.1:4040/metrics/json/ +``` + +These endpoints returned application metadata and metrics in the local kind +executor test. The UI and event logs can expose SQL, configuration and data; +restrict access and do not expose them directly on an untrusted network. + +To retain Spark event logs, configure `spark.eventLog.enabled=true` and +`spark.eventLog.dir` in `spark.extraConf` with a pre-created, writable durable +filesystem location. A separately managed Spark History Server can read the same +location through `spark.history.fs.logDirectory`; its storage credentials and +access controls must be configured separately. In a single-node kind experiment, +a test PVC replaced the driver's `/tmp` scratch volume, with +`spark.eventLog.dir=file:/tmp/events`. A separate History Server mounted the same +PVC and used `spark.history.fs.logDirectory=file:/eventlogs/events`. After a +query and driver restart, its API listed the completed application and its +successful jobs. This manual PVC configuration is not provisioned by the chart and +does not validate multi-node volume sharing or production access controls. See +[Spark monitoring documentation](https://spark.apache.org/docs/4.1.3/monitoring.html) +for metrics sinks and history server details. + +Spark driver logs include Connect session IDs and Spark job details; Hudi SQL +`CALL show_commits` exposes completed instants. Use timestamps and table names to +investigate a particular write, but this setup does not provide a guaranteed +client-request-to-job-to-commit identifier or complete trace correlation. + +### Client-disconnect experiment + +For cancellation, disconnect and lost-response experiments, `--wait` bounds +readback only; process cleanup can take four additional seconds. Expiry is +inconclusive and retains test data; it does not prove the server query stopped. + +On a disposable endpoint, the following creates a unique database and terminates +its own ADBC client during an INSERT. It does not retry the write: + +```bash +python hudi-lakehouse/local-dev/example/connect_disconnect.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --rows 2000 --delay 1 --wait 90 +``` + +If the disconnect client receives a success response before or during interruption, +the result is inconclusive. Later readback proves data is visible, not when it committed. + +If the complete expected write is observed after refreshing the table, the script +cleans up its own table and database. Otherwise it reports an inconclusive outcome +and leaves them for inspection. Wait until the server-side write has stopped before +manually dropping those resources. On a local Spark 4.1.3 / Hudi 1.2.0 endpoint, +a 20,000-row write was observed after the client was terminated one second after +entering the write. This timing does not guarantee the same outcome in another run. +Earlier inconclusive runs used an invalid INSERT with ambiguous column names and +must not be interpreted as evidence about interrupted writes. + +For a controlled lost-response experiment, expose a disposable unauthenticated +endpoint only on localhost and run: + +```bash +python hudi-lakehouse/local-dev/example/connect_lost_response.py --port 15002 +``` + +A local TCP proxy forwards client requests while discarding server-to-client bytes +after the connection is established. A separate connection refreshes and checks the +unique table; once all expected rows are visible, the script terminates the client +and cleans up. A local 1,000-row run observed the complete commit while responses +were blocked. With the optional `--retry` flag, the same INSERT is repeated only +after the original commit is observed; in that run it produced 2,000 rows with +1,000 distinct IDs. This shows an ambiguous client outcome can coexist with a committed +write; it does not establish cancellation, automatic retries, idempotency, or behavior +at every failure phase. If the commit is not observed, the script leaves the table +for inspection after the server write stops. + +### Result-size experiment + +The optional `connect_load.py` reads synthetic `range` rows without creating a +table. Run it only on a disposable endpoint and monitor driver and client memory: + +```bash +python hudi-lakehouse/local-dev/example/connect_load.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --rows 1000000 --payload-bytes 64 --fetch-size 10000 +``` + +In a local Spark 4.1.3/Hudi 1.2.0 test with a 2 GiB driver heap, +`spark.connect.grpc.arrow.maxBatchSize=1m`, Foundry 0.2.1 and Python 3.11 on +macOS, one million rows each containing an int64 ID and 64-byte ASCII payload +passed count and checksum validation using `fetchmany(10000)`. Client peak RSS +was about 170 MiB; a sampled Spark container value was 2.15 GiB, not a peak. +The standalone container had a 10.2 GiB memory limit; this run does not establish +behaviour under the chart's 3 GiB pod limit. A single 5,000,000-byte string row +failed with Foundry's 4,194,304-byte gRPC receive limit despite the 1 MiB Arrow +batch setting. These are observations, not total-result or memory guarantees. + +The opt-in dialect live test also checks Spark `TIMESTAMP_NTZ` with microsecond +precision, decimal(38,18), nullable array elements in a direct SQL result, nested +structs and Hudi commit/key metadata columns. A direct SQL array result does not +establish that Hudi/Parquet can write arrays with NULL elements. In a separate +local CoW table test, an array containing NULL failed during Hudi write; the +server log reported Parquet Avro's old-list-structure null-element restriction. +Setting `spark.hadoop.parquet.avro.write-old-list-structure=false` in the SQL +session did not resolve the failure. Supplying the same property at driver startup +allowed a local CoW INSERT containing `[1, NULL, 3]` to succeed and read back +correctly. In Helm, add the following under `spark.extraConf` before restarting +the endpoint; test the resulting Parquet representation with your readers: Review Comment: I'd avoid recommending this as chart guidance: - **It's global.** `write-old-list-structure=false` at driver startup switches the Parquet list encoding for *every* table this endpoint writes. It isn't limited to tables with nullable arrays. - **It mixes layouts within a table.** Other writers of the same table, such as the Spark 3.5 writer fixture in this PR, Flink or the streamer, keep the default old layout. A single table ends up with different list layouts in different file groups, and compaction and clustering then read one layout and write the other. I haven't verified how every reader handles that, and "test with your readers" pushes that work onto users. Plan item M4.2 was to triage this and, if it's a Hudi write bug, file it separately. Could we file the issue, link it here, and present it as a known limitation ("arrays with NULL elements can't be written through the endpoint yet, see #xxxx") instead of a config recipe? Keeping `connect_nullable_array.py` as a reproducer is useful. ########## hudi-lakehouse/images/spark-connect/check-ready.py: ########## @@ -0,0 +1,64 @@ +# 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. +"""Read-only, opt-in SQL/catalog/table-data readiness for Spark Connect.""" +import argparse +import re +import sys + +from pyspark.sql import SparkSession + + +def identifier(value): + if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', value): + raise argparse.ArgumentTypeError('invalid SQL identifier') + return value + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--schema', required=True, type=identifier) + parser.add_argument('--table', required=True, type=identifier) + parser.add_argument('--column', required=True, type=identifier) + args = parser.parse_args() + spark = None + stage = 'connection' + try: + spark = SparkSession.builder.remote('sc://127.0.0.1:15002').getOrCreate() + stage = 'sql' + if spark.sql('SELECT 1').collect()[0][0] != 1: + raise ValueError('unexpected SQL result') + print('PASS: sql', flush=True) + stage = 'catalog' + if not any(row[0] == args.schema for row in spark.sql('SHOW DATABASES').collect()): + raise ValueError('catalog schema not found') + print('PASS: catalog', flush=True) + stage = 'storage' + rows = spark.sql(f'SELECT `{args.column}` FROM `{args.schema}`.`{args.table}` ' + f'WHERE `{args.column}` IS NOT NULL LIMIT 1').collect() Review Comment: Two operational concerns with the probe: 1. **It goes unready under load.** `SELECT 1` and `SHOW DATABASES` run on the driver, but this scan needs a task slot. With `local[N]` and the FIFO scheduler, a large user query that holds every slot makes the probe wait. After `failureThreshold: 3` × 30 s, the only replica is removed from the Service and *new* clients are refused, even though the server is healthy and busy. Please either document this next to the option, or run the probe in its own scheduler pool / with a longer threshold. 2. **It has no deadline of its own.** If HMS or object storage hangs, the script relies on kubelet's exec timeout to stop it. This Python process also runs inside the driver container's memory limit (2g heap in a 4Gi limit). A deadline inside the script, shorter than `timeoutSeconds` (for example `signal.alarm(25)` or a gRPC deadline), would make sure a stuck probe can't pile up. ########## hudi-lakehouse/charts/hudi-spark-connect/README.md: ########## @@ -346,6 +488,182 @@ to complete. Set the same property in `spark.extraConf` for other deployments us this driver. It controls batch size, not total result size or memory; oversized individual rows can still exceed the receive limit. +### SQL, catalog and storage diagnostics + +By default, Kubernetes readiness checks only the TCP listener. To opt into +SQL-aware readiness, first provision a persistent non-empty storage-backed probe +table with a non-null ordinary data column and use values like: + +```yaml +readiness: + sql: + enabled: true + schema: probe_database + table: probe_table + column: probe_column +``` + +The driver image must include `check-ready.py`; rebuild older images. Startup +continues to use TCP so the driver can initialize before the SQL probe runs. +Readiness checks SQL execution, catalog discovery and an actual table-data query +in sequence. A missing or empty probe table makes the pod unready; provision it +before enabling the option. Checks run every 30 seconds with a 30-second timeout. +In the local kind executor test, the probe passed against a pre-populated Hudi +probe table; changing the configured column to a nonexistent name made the pod +unready. Caching and catalog configuration can affect which backing services a +successful read exercises; this check does not prove write access. + +For a separate read-only data-path diagnostic using ADBC, run: + +```bash +python hudi-lakehouse/local-dev/example/connect_readiness.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --schema probe_database --table probe_table --column probe_column +``` + +The command reports SQL, catalog and table-data checks separately and exits +nonzero at the first failure. It requires read access and never modifies the probe +table. A missing/empty table is a failure; caching and catalog configuration can +affect which backing services a successful read actually exercises. This external +diagnostic does not itself change Kubernetes readiness or prove write access. + +### Spark UI, logs and metrics + +For local diagnosis, forward the driver UI to loopback only: + +```bash +kubectl -n hudi-lakehouse port-forward --address 127.0.0.1 deployment/hudi-spark-connect 4040:4040 +curl http://127.0.0.1:4040/api/v1/applications +curl http://127.0.0.1:4040/metrics/json/ +``` + +These endpoints returned application metadata and metrics in the local kind +executor test. The UI and event logs can expose SQL, configuration and data; +restrict access and do not expose them directly on an untrusted network. + +To retain Spark event logs, configure `spark.eventLog.enabled=true` and +`spark.eventLog.dir` in `spark.extraConf` with a pre-created, writable durable +filesystem location. A separately managed Spark History Server can read the same +location through `spark.history.fs.logDirectory`; its storage credentials and +access controls must be configured separately. In a single-node kind experiment, +a test PVC replaced the driver's `/tmp` scratch volume, with +`spark.eventLog.dir=file:/tmp/events`. A separate History Server mounted the same +PVC and used `spark.history.fs.logDirectory=file:/eventlogs/events`. After a +query and driver restart, its API listed the completed application and its +successful jobs. This manual PVC configuration is not provisioned by the chart and +does not validate multi-node volume sharing or production access controls. See +[Spark monitoring documentation](https://spark.apache.org/docs/4.1.3/monitoring.html) +for metrics sinks and history server details. + +Spark driver logs include Connect session IDs and Spark job details; Hudi SQL +`CALL show_commits` exposes completed instants. Use timestamps and table names to +investigate a particular write, but this setup does not provide a guaranteed +client-request-to-job-to-commit identifier or complete trace correlation. + +### Client-disconnect experiment + +For cancellation, disconnect and lost-response experiments, `--wait` bounds +readback only; process cleanup can take four additional seconds. Expiry is +inconclusive and retains test data; it does not prove the server query stopped. + +On a disposable endpoint, the following creates a unique database and terminates +its own ADBC client during an INSERT. It does not retry the write: + +```bash +python hudi-lakehouse/local-dev/example/connect_disconnect.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --rows 2000 --delay 1 --wait 90 +``` + +If the disconnect client receives a success response before or during interruption, +the result is inconclusive. Later readback proves data is visible, not when it committed. + +If the complete expected write is observed after refreshing the table, the script +cleans up its own table and database. Otherwise it reports an inconclusive outcome +and leaves them for inspection. Wait until the server-side write has stopped before +manually dropping those resources. On a local Spark 4.1.3 / Hudi 1.2.0 endpoint, +a 20,000-row write was observed after the client was terminated one second after +entering the write. This timing does not guarantee the same outcome in another run. +Earlier inconclusive runs used an invalid INSERT with ambiguous column names and +must not be interpreted as evidence about interrupted writes. + +For a controlled lost-response experiment, expose a disposable unauthenticated +endpoint only on localhost and run: + +```bash +python hudi-lakehouse/local-dev/example/connect_lost_response.py --port 15002 +``` + +A local TCP proxy forwards client requests while discarding server-to-client bytes +after the connection is established. A separate connection refreshes and checks the +unique table; once all expected rows are visible, the script terminates the client +and cleans up. A local 1,000-row run observed the complete commit while responses +were blocked. With the optional `--retry` flag, the same INSERT is repeated only +after the original commit is observed; in that run it produced 2,000 rows with +1,000 distinct IDs. This shows an ambiguous client outcome can coexist with a committed +write; it does not establish cancellation, automatic retries, idempotency, or behavior +at every failure phase. If the commit is not observed, the script leaves the table +for inspection after the server write stops. + +### Result-size experiment + +The optional `connect_load.py` reads synthetic `range` rows without creating a +table. Run it only on a disposable endpoint and monitor driver and client memory: + +```bash +python hudi-lakehouse/local-dev/example/connect_load.py \ + --uri 'spark://localhost:15002?api=connect&auth_type=none&tls=false' \ + --rows 1000000 --payload-bytes 64 --fetch-size 10000 +``` + +In a local Spark 4.1.3/Hudi 1.2.0 test with a 2 GiB driver heap, +`spark.connect.grpc.arrow.maxBatchSize=1m`, Foundry 0.2.1 and Python 3.11 on +macOS, one million rows each containing an int64 ID and 64-byte ASCII payload +passed count and checksum validation using `fetchmany(10000)`. Client peak RSS +was about 170 MiB; a sampled Spark container value was 2.15 GiB, not a peak. +The standalone container had a 10.2 GiB memory limit; this run does not establish +behaviour under the chart's 3 GiB pod limit. A single 5,000,000-byte string row Review Comment: Two small fixes: - **Pod limit:** the chart sets `requests.memory: 3Gi` and `limits.memory: 4Gi`, so "the chart's 3 GiB pod limit" should say 4 GiB, or "3 GiB request". - **macOS (line 620):** how was Foundry 0.2.1 installed on macOS? In my local run of the #20467 quickstart, `dbc install 'spark=0.2.1'` on a macOS arm64 host failed with "no package found", so I ran all clients from Linux pods. If there's a working macOS route, please document it. If not, the README should note that the host-side driver needs Linux. ########## hudi-lakehouse/local-dev/example/connect_cancel.py: ########## @@ -0,0 +1,111 @@ +# 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. +"""Opt-in ADBC cancellation experiment against a disposable localhost endpoint.""" +import argparse +import os +import threading +import time +import uuid + +from connect_fault_utils import readback + + +def connect(port, driver): + from adbc_driver_manager import dbapi + uri = f'spark://127.0.0.1:{port}?api=connect&auth_type=none&tls=false' + return dbapi.connect(driver=driver, db_kwargs={'uri': uri}, autocommit=True) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument('--port', type=int, required=True) + p.add_argument('--driver', default='spark') + p.add_argument('--rows', type=int, default=20000) + p.add_argument('--delay', type=float, default=1) + p.add_argument('--wait', type=int, default=90) + args = p.parse_args() + if not 1 <= args.port <= 65535 or args.rows < 1 or args.delay < 0 or args.wait < 1: + p.error('invalid port, row count, delay or wait') + database = 'connect_cancel_' + uuid.uuid4().hex + table = database + '.orders' + print(f'Test database: {database}', flush=True) + created = False + with connect(args.port, args.driver) as observer, observer.cursor() as read, \ + connect(args.port, args.driver) as writer, writer.cursor() as write: + try: + read.execute(f'CREATE DATABASE {database}') + created = True + read.execute(f"CREATE TABLE {table} (id INT, ts BIGINT) USING hudi " + "TBLPROPERTIES(type='cow', primaryKey='id', preCombineField='ts')") + outcome = [] + started = threading.Event() + + def submit(): + started.set() + try: + write.execute(f'INSERT INTO {table} SELECT CAST(id AS INT) AS id, CAST(id AS BIGINT) AS ts FROM range({args.rows})') + outcome.append('success') + except Exception as exc: + outcome.append(f'{type(exc).__name__}: {exc}') + + worker = threading.Thread(target=submit, daemon=True) + worker.start() + started.wait() + time.sleep(args.delay) + if not worker.is_alive(): + print('INCONCLUSIVE: write finished before cancellation', flush=True) + return 2 + try: + write.adbc_cancel() + print('Cancellation request returned', flush=True) + except Exception as exc: + print(f'Cancellation request failed: {type(exc).__name__}: {exc}', flush=True) + worker.join(30) + if worker.is_alive(): + # Do not close a cursor concurrently with an active operation. + print(f'INCONCLUSIVE: client did not stop; leave {database} for inspection', flush=True) + os._exit(2) + print(f'Client write result: {outcome}', flush=True) + deadline = time.monotonic() + args.wait + while time.monotonic() < deadline: + try: + count, total = readback( + f'spark://127.0.0.1:{args.port}?api=connect&auth_type=none&tls=false', + args.driver, table, deadline) + except TimeoutError: + print('INCONCLUSIVE: readback deadline expired; preserve test data', flush=True) + return 2 + except Exception as exc: + print(f'Readback not yet available: {type(exc).__name__}', flush=True) + time.sleep(min(2, max(0, deadline - time.monotonic()))) + continue + if count == args.rows and total == args.rows * (args.rows - 1) // 2: + print(f'OBSERVED: all {count} rows visible after the cancellation attempt; commit timing is unknown', flush=True) + read.execute(f'DROP TABLE IF EXISTS {table} PURGE') + read.execute(f'DROP DATABASE IF EXISTS {database}') + created = False + print('PASS: owned test data cleaned', flush=True) + return 0 + time.sleep(min(2, max(0, deadline - time.monotonic()))) + print('INCONCLUSIVE: complete commit was not observed; outcome may still change', flush=True) Review Comment: This experiment can only PASS when every row committed, meaning the cancel failed to stop the write. If `adbc_cancel()` really aborts the Spark job, the table stays empty, the loop runs until `--wait` expires, and the script reports INCONCLUSIVE. So the outcome we'd actually want can't be observed. Could the observer also check the timeline (for example `hudi_query_timeline('<table>', 'false')`)? If no commit was completed for this write and its instant is rolled back, or only an inflight instant is left after the client returned an error, report `OBSERVED: write aborted` with 0 rows. That gives M2.3/M4 an observable answer either way. -- 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]
