yuqi1129 commented on code in PR #11266: URL: https://github.com/apache/gravitino/pull/11266#discussion_r3321967829
########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/purge/IcebergPurgeJobSQLProviderFactory.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.iceberg.service.purge; + +import static org.apache.gravitino.iceberg.service.purge.IcebergPurgeJobMapper.TABLE_NAME; + +import com.google.common.collect.ImmutableMap; +import java.util.Map; +import org.apache.gravitino.storage.relational.JDBCBackend.JDBCBackendType; +import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.apache.ibatis.annotations.Param; + +/** + * Supplies the SQL for {@link IcebergPurgeJobMapper}, dispatching by the entity store's configured + * JDBC backend. The statements are written with portable, parameterized SQL (the row id and all + * timestamps are generated by the application, not the database), so the {@link BaseProvider} + * serves MySQL, H2, and PostgreSQL alike. A backend that ever needs a divergent statement can be + * registered here by mapping its {@link JDBCBackendType} to a {@code BaseProvider} subclass that + * overrides only the affected methods. + */ +public class IcebergPurgeJobSQLProviderFactory { + + private static final BaseProvider BASE_PROVIDER = new BaseProvider(); + + private static final Map<JDBCBackendType, BaseProvider> PROVIDERS = + ImmutableMap.of( + JDBCBackendType.MYSQL, BASE_PROVIDER, + JDBCBackendType.H2, BASE_PROVIDER, + JDBCBackendType.POSTGRESQL, BASE_PROVIDER); + + private static BaseProvider getProvider() { + String databaseId = + SqlSessionFactoryHelper.getInstance() + .getSqlSessionFactory() + .getConfiguration() + .getDatabaseId(); + return PROVIDERS.get(JDBCBackendType.fromString(databaseId)); + } + + public static String insertPurgeJob(@Param("po") IcebergPurgeJobPO po) { + return getProvider().insertPurgeJob(po); + } + + public static String selectClaimableIds( + @Param("staleBefore") long staleBefore, @Param("window") int window) { + return getProvider().selectClaimableIds(staleBefore, window); + } + + public static String markRunning( + @Param("id") long id, @Param("now") long now, @Param("staleBefore") long staleBefore) { + return getProvider().markRunning(id, now, staleBefore); + } + + public static String selectById(@Param("id") long id) { + return getProvider().selectById(id); + } + + public static String markSucceeded(@Param("id") long id, @Param("now") long now) { + return getProvider().markSucceeded(id, now); + } + + public static String markFailed( + @Param("id") long id, @Param("reason") String reason, @Param("now") long now) { + return getProvider().markFailed(id, reason, now); + } + + public static String recordFailure( + @Param("id") long id, + @Param("reason") String reason, + @Param("maxAttempts") int maxAttempts, + @Param("now") long now) { + return getProvider().recordFailure(id, reason, maxAttempts, now); + } + + public static String heartbeat( + @Param("id") long id, @Param("lastWritten") long lastWritten, @Param("now") long now) { + return getProvider().heartbeat(id, lastWritten, now); + } + + public static String selectActiveJobId( + @Param("catalog") String catalog, + @Param("namespace") String namespace, + @Param("table") String table) { + return getProvider().selectActiveJobId(catalog, namespace, table); + } + + public static String pruneFinishedBefore(@Param("updatedBefore") long updatedBefore) { + return getProvider().pruneFinishedBefore(updatedBefore); + } + + public static String selectState(@Param("id") long id) { + return getProvider().selectState(id); + } + + /** Portable SQL shared by all supported JDBC backends. */ + static class BaseProvider { + + String insertPurgeJob(@Param("po") IcebergPurgeJobPO po) { + return "INSERT INTO " + + TABLE_NAME + + " (id, metalake_name, catalog_name, namespace, table_name, metadata_location," + + " file_io_impl, file_io_props, state, attempts, last_error, heartbeat_at, created_by," + + " updated_at) VALUES (#{po.id}, #{po.metalakeName}, #{po.catalogName}, #{po.namespace}," + + " #{po.tableName}, #{po.metadataLocation}, #{po.fileIOImpl}, #{po.fileIOProps}," + + " #{po.state}, #{po.attempts}, #{po.lastError}, #{po.heartbeatAt}, #{po.createdBy}," + + " #{po.updatedAt})"; + } + + String selectClaimableIds(@Param("staleBefore") long staleBefore, @Param("window") int window) { + return "SELECT id FROM " + + TABLE_NAME + + " WHERE state = 'PENDING'" + + " OR (state = 'RUNNING' AND (heartbeat_at IS NULL OR heartbeat_at < #{staleBefore}))" Review Comment: Can we set the default value to 0 for `heartbeat_at`? `OR` will skip the index make and query not very efficient. -- 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]
