-
Notifications
You must be signed in to change notification settings - Fork 276
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rbroughan/stream status persistence server (#6235)
- Loading branch information
Showing
10 changed files
with
665 additions
and
8 deletions.
There are no files selected for viewing
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
119 changes: 119 additions & 0 deletions
119
airbyte-server/src/main/java/io/airbyte/server/repositories/StreamStatusRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.repositories; | ||
|
|
||
| import static io.airbyte.db.instance.jobs.jooq.generated.Tables.STREAM_STATUSES; | ||
|
|
||
| import com.google.common.base.CaseFormat; | ||
| import io.airbyte.server.repositories.domain.StreamStatus; | ||
| import io.micronaut.data.annotation.Repository; | ||
| import io.micronaut.data.annotation.RepositoryConfiguration; | ||
| import io.micronaut.data.model.Page; | ||
| import io.micronaut.data.model.Pageable; | ||
| import io.micronaut.data.model.query.builder.jpa.JpaQueryBuilder; | ||
| import io.micronaut.data.repository.PageableRepository; | ||
| import io.micronaut.data.repository.jpa.JpaSpecificationExecutor; | ||
| import io.micronaut.data.repository.jpa.criteria.PredicateSpecification; | ||
| import java.util.UUID; | ||
| import lombok.Builder; | ||
| import org.jooq.TableField; | ||
|
|
||
| /** | ||
| * Data Access layer for StreamStatus. | ||
| */ | ||
| @SuppressWarnings("MissingJavadocType") | ||
| @Repository | ||
| @RepositoryConfiguration(queryBuilder = JpaQueryBuilder.class) // Despite what your IDE says, we must specify the query builder | ||
| public interface StreamStatusRepository extends PageableRepository<StreamStatus, UUID>, JpaSpecificationExecutor<StreamStatus> { | ||
|
|
||
| /** | ||
| * Returns stream statuses filtered by the provided params. | ||
| */ | ||
| default Page<StreamStatus> findAllFiltered(final FilterParams params) { | ||
| var spec = Predicates.columnEquals(Columns.WORKSPACE_ID, params.workspaceId()); | ||
| var pageable = Pageable.unpaged(); | ||
|
|
||
| if (null != params.connectionId()) { | ||
| spec = spec.and(Predicates.columnEquals(Columns.CONNECTION_ID, params.connectionId())); | ||
| } | ||
|
|
||
| if (null != params.jobId()) { | ||
| spec = spec.and(Predicates.columnEquals(Columns.JOB_ID, params.jobId())); | ||
| } | ||
|
|
||
| if (null != params.streamNamespace()) { | ||
| spec = spec.and(Predicates.columnEquals(Columns.STREAM_NAMESPACE, params.streamNamespace())); | ||
| } | ||
|
|
||
| if (null != params.streamName()) { | ||
| spec = spec.and(Predicates.columnEquals(Columns.STREAM_NAME, params.streamName())); | ||
| } | ||
|
|
||
| if (null != params.attemptNumber()) { | ||
| spec = spec.and(Predicates.columnEquals(Columns.ATTEMPT_NUMBER, params.attemptNumber())); | ||
| } | ||
|
|
||
| if (null != params.pagination()) { | ||
| final var offset = params.pagination().offset(); | ||
| final var size = params.pagination().size(); | ||
| pageable = Pageable.from(offset, size); | ||
| } | ||
|
|
||
| return findAll(spec, pageable); | ||
| } | ||
|
|
||
| /** | ||
| * Pagination params. | ||
| */ | ||
| record Pagination(int offset, | ||
| int size) {} | ||
|
|
||
| /** | ||
| * Params for filtering our list functionality. | ||
| */ | ||
| @Builder | ||
| record FilterParams(UUID workspaceId, | ||
| UUID connectionId, | ||
| Long jobId, | ||
| String streamNamespace, | ||
| String streamName, | ||
| Integer attemptNumber, | ||
| Pagination pagination) {} | ||
|
|
||
| /** | ||
| * Predicates for dynamic query building. Portable. | ||
| */ | ||
| class Predicates { | ||
|
|
||
| /* | ||
| * Jooq holds onto the names of the columns in snake_case, so we have to convert to lower camelCase | ||
| * for Hibernate to do predicate filtering. | ||
| */ | ||
| static String formatJooqColumnName(final TableField<?, ?> jooqColumn) { | ||
| return CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, jooqColumn.getName()); | ||
| } | ||
|
|
||
| static <U> PredicateSpecification<StreamStatus> columnEquals(final String columnName, final U value) { | ||
| return (root, criteriaBuilder) -> criteriaBuilder.equal(root.get(columnName), value); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Column names for StreamStatus in camel case for Hibernate. In lieu of a metamodel, we pre-create | ||
| * Hibernate-friendly column names for the already generated Jooq model. | ||
| */ | ||
| class Columns { | ||
|
|
||
| static String WORKSPACE_ID = Predicates.formatJooqColumnName(STREAM_STATUSES.WORKSPACE_ID); | ||
| static String CONNECTION_ID = Predicates.formatJooqColumnName(STREAM_STATUSES.CONNECTION_ID); | ||
| static String JOB_ID = Predicates.formatJooqColumnName(STREAM_STATUSES.JOB_ID); | ||
| static String STREAM_NAMESPACE = Predicates.formatJooqColumnName(STREAM_STATUSES.STREAM_NAMESPACE); | ||
| static String STREAM_NAME = Predicates.formatJooqColumnName(STREAM_STATUSES.STREAM_NAME); | ||
| static String ATTEMPT_NUMBER = Predicates.formatJooqColumnName(STREAM_STATUSES.ATTEMPT_NUMBER); | ||
|
|
||
| } | ||
|
|
||
| } |
81 changes: 81 additions & 0 deletions
81
airbyte-server/src/main/java/io/airbyte/server/repositories/domain/StreamStatus.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.server.repositories.domain; | ||
|
|
||
| import io.airbyte.db.instance.jobs.jooq.generated.enums.JobStreamStatusIncompleteRunCause; | ||
| import io.airbyte.db.instance.jobs.jooq.generated.enums.JobStreamStatusJobType; | ||
| import io.airbyte.db.instance.jobs.jooq.generated.enums.JobStreamStatusRunState; | ||
| import io.hypersistence.utils.hibernate.type.basic.PostgreSQLEnumType; | ||
| import io.micronaut.data.annotation.DateCreated; | ||
| import io.micronaut.data.annotation.DateUpdated; | ||
| import java.time.OffsetDateTime; | ||
| import java.util.UUID; | ||
| import javax.persistence.Entity; | ||
| import javax.persistence.EnumType; | ||
| import javax.persistence.Enumerated; | ||
| import javax.persistence.GeneratedValue; | ||
| import javax.persistence.Id; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Builder; | ||
| import lombok.EqualsAndHashCode; | ||
| import lombok.Getter; | ||
| import lombok.NoArgsConstructor; | ||
| import lombok.Setter; | ||
| import org.hibernate.annotations.Type; | ||
| import org.hibernate.annotations.TypeDef; | ||
|
|
||
| /** | ||
| * DTO for our data access layer. | ||
| */ | ||
| @Builder | ||
| @AllArgsConstructor // The builder uses this constructor | ||
| @NoArgsConstructor // Hibernate uses this constructor and sets the values individually | ||
| @Getter | ||
| @Setter | ||
| @EqualsAndHashCode(exclude = {"id", "createdAt", "updatedAt"}) | ||
| @Entity(name = "stream_statuses") | ||
| @TypeDef(name = StreamStatus.PGSQL_ENUM, | ||
| typeClass = PostgreSQLEnumType.class) | ||
| public class StreamStatus { | ||
|
|
||
| static final String PGSQL_ENUM = "pgsql_enum"; | ||
|
|
||
| @Id | ||
| @GeneratedValue | ||
| private UUID id; | ||
|
|
||
| private UUID workspaceId; | ||
|
|
||
| private UUID connectionId; | ||
|
|
||
| private Long jobId; | ||
|
|
||
| private Integer attemptNumber; | ||
|
|
||
| private String streamNamespace; | ||
|
|
||
| private String streamName; | ||
|
|
||
| @Enumerated(EnumType.STRING) | ||
| @Type(type = PGSQL_ENUM) | ||
| private JobStreamStatusJobType jobType; | ||
|
|
||
| @DateCreated | ||
| private OffsetDateTime createdAt; | ||
|
|
||
| @DateUpdated | ||
| private OffsetDateTime updatedAt; | ||
|
|
||
| @Enumerated(EnumType.STRING) | ||
| @Type(type = PGSQL_ENUM) | ||
| private JobStreamStatusRunState runState; | ||
|
|
||
| @Enumerated(EnumType.STRING) | ||
| @Type(type = PGSQL_ENUM) | ||
| private JobStreamStatusIncompleteRunCause incompleteRunCause; | ||
|
|
||
| private OffsetDateTime transitionedAt; | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.