lasdf1234 commented on code in PR #13762: URL: https://github.com/apache/gravitino/pull/13762#discussion_r4246565104
########## scripts/h2/schema-2.0.0-h2.sql: ########## @@ -680,3 +680,32 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` ( ) COMMENT='async Iceberg table cleanup jobs'; CREATE INDEX IF NOT EXISTS `idx_state_updated` ON `iceberg_cleanup_job` (`state`, `updated_at`); CREATE INDEX IF NOT EXISTS `idx_object` ON `iceberg_cleanup_job` (`catalog_id`, `namespace`, `table_name`, `state`); + +CREATE TABLE IF NOT EXISTS `table_maintenance_job` ( Review Comment: Agreed — renamed to `table_maintenance_action` (DDL + DAOs / mappers / POs / tests). The row is a TMS-domain action/occupancy record; `job_id` is optional and only bound after `runJob`. ########## plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceJobMetaService.java: ########## @@ -0,0 +1,129 @@ +/* + * 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. + */ +package org.apache.gravitino.maintenance.storage.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; + +import javax.annotation.Nullable; +import org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceJobMapper; +import org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper; +import org.apache.gravitino.maintenance.storage.po.TableMaintenanceJobPO; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** DAO service for {@code table_maintenance_job}. */ +public class TableMaintenanceJobMetaService { + + private static final TableMaintenanceJobMetaService INSTANCE = + new TableMaintenanceJobMetaService(); + + private TableMaintenanceJobMetaService() {} + + /** + * @return singleton service instance + */ + public static TableMaintenanceJobMetaService getInstance() { + return INSTANCE; + } + + /** + * Inserts a maintenance job row with {@code job_id}, {@code table_id}, {@code policy_id}, and + * {@code created_at} (§5.2). + * + * @param jobId {@code job_run_meta.job_run_id} + * @param tableId table entity id + * @param policyId policy entity id + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertTableMaintenanceJob") + public void insertJob(long jobId, long tableId, long policyId) { Review Comment: Fixed. Occupancy is now created via `insertInFlight(tableId, policyId)` with nullable `job_id`, then bound with `bindJobId(id, jobId)` after a successful `runJob`, so the check-to-submit race can be closed before submission. ########## plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableSnapshotMetricsBaseSQLProvider.java: ########## @@ -0,0 +1,81 @@ +/* + * 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. + */ +package org.apache.gravitino.maintenance.storage.mapper.provider.base; + +import org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceJobMapper; +import org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper; +import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO; +import org.apache.ibatis.annotations.Param; + +/** Portable SQL for {@code table_snapshot_metrics}. */ +public class TableSnapshotMetricsBaseSQLProvider { + + /** + * @param metricsMeta metrics row + * @return INSERT statement + */ + public String insertMetrics(@Param("metricsMeta") TableSnapshotMetricsPO metricsMeta) { + return "INSERT INTO " + + TableSnapshotMetricsMapper.TABLE_NAME + + " (table_id, policy_id, snapshot_id, metrics_value, created_at)" + + " VALUES (#{metricsMeta.tableId}, #{metricsMeta.policyId}, #{metricsMeta.snapshotId}," + + " #{metricsMeta.metricsValue}, #{metricsMeta.createdAt})"; + } + + /** + * @param tableId table id + * @param policyId policy id + * @param snapshotId Iceberg snapshot id + * @return SELECT statement + */ + public String selectMetrics( + @Param("tableId") long tableId, + @Param("policyId") long policyId, + @Param("snapshotId") long snapshotId) { + return "SELECT id, table_id AS tableId, policy_id AS policyId, snapshot_id AS snapshotId," + + " metrics_value AS metricsValue, created_at AS createdAt FROM " + + TableSnapshotMetricsMapper.TABLE_NAME + + " WHERE table_id = #{tableId} AND policy_id = #{policyId} AND snapshot_id =" + + " #{snapshotId}"; + } + + /** + * Deletes metrics rows tied to a maintenance job's before/after snapshot ids. + * + * @param jobId Gravitino job run id + * @return DELETE statement + */ + public String deleteByJobId(@Param("jobId") long jobId) { Review Comment: Fixed. Action deletion no longer cascades into `table_snapshot_metrics`; metrics are retained for TTL/N cleanup as designed. ########## plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceJobBaseSQLProvider.java: ########## @@ -0,0 +1,97 @@ +/* + * 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. + */ +package org.apache.gravitino.maintenance.storage.mapper.provider.base; + +import static org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceJobMapper.TABLE_NAME; + +import org.apache.gravitino.maintenance.storage.po.TableMaintenanceJobPO; +import org.apache.ibatis.annotations.Param; + +/** Portable SQL for {@code table_maintenance_job}. */ +public class TableMaintenanceJobBaseSQLProvider { + + /** + * Inserts a run row with {@code job_id}, {@code table_id}, {@code policy_id}, and {@code + * created_at}. Snapshot columns stay null until the Spark Job / terminal listener updates them + * (§5.2). + * + * @param jobMeta row to insert + * @return INSERT statement + */ + public String insertJob(@Param("jobMeta") TableMaintenanceJobPO jobMeta) { + return "INSERT INTO " + + TABLE_NAME + + " (job_id, table_id, policy_id, created_at)" + + " VALUES (#{jobMeta.jobId}, #{jobMeta.tableId}, #{jobMeta.policyId}," + + " #{jobMeta.createdAt})"; + } + + /** + * @param jobId Gravitino job run id + * @return SELECT statement + */ + public String selectByJobId(@Param("jobId") long jobId) { + return "SELECT id, job_id AS jobId, table_id AS tableId, policy_id AS policyId," + + " before_snapshot_id AS beforeSnapshotId, after_snapshot_id AS afterSnapshotId," + + " finished_at AS finishedAt, created_at AS createdAt FROM " + + TABLE_NAME + + " WHERE job_id = #{jobId}"; + } + + /** + * Sets {@code before_snapshot_id} from inside the Spark Job when {@code S0} is sampled. + * + * @param jobId Gravitino job run id + * @param beforeSnapshotId sampled snapshot id + * @return UPDATE statement + */ + public String updateBeforeSnapshot( + @Param("jobId") long jobId, @Param("beforeSnapshotId") long beforeSnapshotId) { + return "UPDATE " + + TABLE_NAME + + " SET before_snapshot_id = #{beforeSnapshotId}" + + " WHERE job_id = #{jobId} AND finished_at IS NULL"; Review Comment: Fixed. `updateBeforeSnapshot` now requires `before_snapshot_id IS NULL AND finished_at IS NULL`, and both `recordBeforeSnapshot` / `markFinished` check affected-row count (`updated == 1`) so duplicate/late callbacks fail explicitly. ########## plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableSnapshotMetricsMetaService.java: ########## @@ -0,0 +1,83 @@ +/* + * 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. + */ +package org.apache.gravitino.maintenance.storage.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; + +import javax.annotation.Nullable; +import org.apache.gravitino.maintenance.storage.mapper.TableSnapshotMetricsMapper; +import org.apache.gravitino.maintenance.storage.po.TableSnapshotMetricsPO; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** DAO service for {@code table_snapshot_metrics}. */ +public class TableSnapshotMetricsMetaService { + + private static final TableSnapshotMetricsMetaService INSTANCE = + new TableSnapshotMetricsMetaService(); + + private TableSnapshotMetricsMetaService() {} + + /** + * @return singleton service instance + */ + public static TableSnapshotMetricsMetaService getInstance() { + return INSTANCE; + } + + /** + * Inserts metrics JSON for a table/policy/snapshot triple. Does not update an existing row. + * + * @param tableId table id + * @param policyId policy id + * @param snapshotId Iceberg snapshot id + * @param metricsJson serialized metrics payload + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertTableSnapshotMetrics") + public void insertMetrics(long tableId, long policyId, long snapshotId, String metricsJson) { + TableSnapshotMetricsPO metricsMeta = + TableSnapshotMetricsPO.builder() + .withTableId(tableId) + .withPolicyId(policyId) + .withSnapshotId(snapshotId) + .withMetricsValue(metricsJson) + .withCreatedAt(System.currentTimeMillis()) + .build(); + SessionUtils.doWithCommit( + TableSnapshotMetricsMapper.class, mapper -> mapper.insertMetrics(metricsMeta)); Review Comment: Fixed. `insertMetrics` is idempotent: if a row for `(table, policy, snapshot)` already exists, the call is a no-op (no overwrite / no duplicate-key failure on retry). -- 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]
