| /* |
| * 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.lens.cube.parse; |
| |
| import static java.util.Comparator.naturalOrder; |
| |
| import static org.apache.lens.cube.parse.CandidateTablePruneCause.CandidateTablePruneCode; |
| import static org.apache.lens.cube.parse.CandidateTablePruneCause.SkipUpdatePeriodCode; |
| import static org.apache.lens.cube.parse.CandidateTablePruneCause.timeDimNotSupported; |
| import static org.apache.lens.cube.parse.StorageUtil.getFallbackRange; |
| import static org.apache.lens.cube.parse.StorageUtil.processCubeColForDataCompleteness; |
| import static org.apache.lens.cube.parse.StorageUtil.processExpressionsForCompleteness; |
| |
| import java.text.DateFormat; |
| import java.text.SimpleDateFormat; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.LinkedHashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.OptionalDouble; |
| import java.util.Set; |
| import java.util.TimeZone; |
| import java.util.TreeSet; |
| import java.util.stream.Stream; |
| |
| import org.apache.lens.cube.metadata.*; |
| import org.apache.lens.server.api.error.LensException; |
| import org.apache.lens.server.api.metastore.DataCompletenessChecker; |
| |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.hadoop.hive.ql.metadata.HiveException; |
| import org.apache.hadoop.hive.ql.session.SessionState; |
| |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import lombok.Getter; |
| import lombok.Setter; |
| import lombok.extern.slf4j.Slf4j; |
| |
| /** |
| * Represents a fact on a storage table and the dimensions it needs to be joined with to answer the query |
| */ |
| @Slf4j |
| public class StorageCandidate implements Candidate, CandidateTable { |
| |
| // TODO union : Put comments on member variables. |
| @Getter |
| private final CubeQueryContext cubeQueryContext; |
| private final String processTimePartCol; |
| private final String completenessPartCol; |
| private final float completenessThreshold; |
| |
| /** |
| * Name of this storage candidate = storageName_factName |
| */ |
| @Getter |
| @Setter |
| private String name; |
| |
| /** |
| * This is the storage table specific name. It is used while generating query from this candidate |
| */ |
| @Setter |
| private String resolvedName; |
| /** |
| * Valid update periods populated by Phase 1. |
| */ |
| @Getter |
| private TreeSet<UpdatePeriod> validUpdatePeriods = new TreeSet<>(); |
| |
| /** |
| * These are the update periods that finally participate in partitions. |
| * @see #getParticipatingPartitions() |
| */ |
| @Getter |
| private TreeSet<UpdatePeriod> participatingUpdatePeriods = new TreeSet<>(); |
| |
| @Getter |
| @Setter |
| Map<String, SkipUpdatePeriodCode> updatePeriodRejectionCause; |
| @Getter |
| Set<Dimension> queriedDims = Sets.newHashSet(); |
| private Collection<StorageCandidate> periodSpecificStorageCandidates; |
| |
| /** |
| * This map holds Tags (A tag refers to one or more measures) that have incomplete (below configured threshold) data. |
| * Value is a map of date string and %completeness. |
| */ |
| @Getter |
| private Map<String, Map<String, Float>> dataCompletenessMap = new HashMap<>(); |
| private SimpleDateFormat partWhereClauseFormat = null; |
| /** |
| * Participating fact, storage and dimensions for this StorageCandidate |
| */ |
| @Getter |
| private FactTable fact; |
| @Getter |
| private String storageName; |
| @Getter |
| private String storageTable; |
| @Getter |
| private Map<TimeRange, Set<FactPartition>> rangeToPartitions = new LinkedHashMap<>(); |
| @Getter |
| private Map<TimeRange, String> rangeToExtraWhereFallBack = new LinkedHashMap<>(); |
| @Getter |
| private Set<Integer> answerableMeasurePhraseIndices = Sets.newHashSet(); |
| @Getter |
| @Setter |
| private String fromString; |
| @Getter |
| private CubeInterface cube; |
| @Getter |
| private Date startTime; |
| @Getter |
| private Date endTime; |
| /** |
| * Cached fact columns |
| */ |
| private Collection<String> factColumns; |
| |
| /** |
| * Non existing partitions |
| */ |
| @Getter |
| private Set<String> nonExistingPartitions = new HashSet<>(); |
| |
| /** |
| * This will be true if this storage candidate has multiple storage tables (one per update period) |
| * https://issues.apache.org/jira/browse/LENS-1386 |
| */ |
| @Getter |
| private boolean isStorageTblsAtUpdatePeriodLevel; |
| |
| @Getter |
| private int numQueriedParts = 0; |
| |
| public StorageCandidate(StorageCandidate sc) throws LensException { |
| this(sc.getCube(), sc.getFact(), sc.getStorageName(), sc.getCubeQueryContext()); |
| this.validUpdatePeriods.addAll(sc.getValidUpdatePeriods()); |
| this.fromString = sc.fromString; |
| this.factColumns = sc.factColumns; |
| this.answerableMeasurePhraseIndices.addAll(sc.answerableMeasurePhraseIndices); |
| for (Map.Entry<TimeRange, Set<FactPartition>> entry : sc.getRangeToPartitions().entrySet()) { |
| rangeToPartitions.put(entry.getKey(), new LinkedHashSet<>(entry.getValue())); |
| } |
| this.rangeToExtraWhereFallBack = sc.rangeToExtraWhereFallBack; |
| } |
| |
| public StorageCandidate(CubeInterface cube, FactTable fact, String storageName, CubeQueryContext cubeQueryContext) |
| throws LensException { |
| this.cube = cube; |
| this.fact = fact; |
| this.cubeQueryContext = cubeQueryContext; |
| if ((getCube() == null) || (fact == null) || (storageName == null)) { |
| throw new IllegalArgumentException("Cube,fact and storageName should be non null"); |
| } |
| this.storageName = storageName; |
| this.storageTable = MetastoreUtil.getFactOrDimtableStorageTableName(fact.getSourceFactName(), storageName); |
| this.name = getFact().getName(); |
| this.processTimePartCol = getConf().get(CubeQueryConfUtil.PROCESS_TIME_PART_COL); |
| String formatStr = getConf().get(CubeQueryConfUtil.PART_WHERE_CLAUSE_DATE_FORMAT); |
| if (formatStr != null) { |
| this.partWhereClauseFormat = new SimpleDateFormat(formatStr); |
| } |
| completenessPartCol = getConf().get(CubeQueryConfUtil.COMPLETENESS_CHECK_PART_COL); |
| completenessThreshold = getConf() |
| .getFloat(CubeQueryConfUtil.COMPLETENESS_THRESHOLD, CubeQueryConfUtil.DEFAULT_COMPLETENESS_THRESHOLD); |
| |
| Set<String> storageTblNames = getCubeMetastoreClient().getStorageTables(fact, storageName); |
| isStorageTblsAtUpdatePeriodLevel = storageTblNames.size() > 1 |
| || !storageTblNames.iterator().next().equalsIgnoreCase(storageTable); |
| setStorageStartAndEndDate(); |
| } |
| |
| String getTimeRangeWhereClasue(TimeRangeWriter rangeWriter, TimeRange range) |
| throws LensException { |
| String rangeWhere = rangeWriter.getTimeRangeWhereClause( |
| getCubeQueryContext(), getCubeQueryContext().getAliasForTableName(getCube().getName()), |
| getRangeToPartitions().get(range)); |
| String fallback = getRangeToExtraWhereFallBack().get(range); |
| if (StringUtils.isNotBlank(fallback)){ |
| rangeWhere = "((" + rangeWhere + ") and (" + fallback + "))"; |
| } |
| return rangeWhere; |
| } |
| |
| /** |
| * Sets Storage candidates start and end time based on underlying storage-tables |
| * |
| * CASE 1 |
| * If has Storage has single storage table* |
| * Storage start time = max(storage start time , fact start time) |
| * Storage end time = min(storage end time , fact start time) |
| * |
| * CASE 2 |
| * If the Storage has multiple Storage Tables (one per update period)* |
| * update Period start Time = Max(update start time, fact start time) |
| * update Period end Time = Min(update end time, fact end time) |
| * Stoarge start and end time is derived form the underlying update period start and end times. |
| * Storage start time = min(update1 start time ,...., updateN start time) |
| * Storage end time = max(update1 end time ,...., updateN end time) |
| * |
| * Note in Case 2 its assumed that the time range supported by different update periods are either |
| * overlapping(Example 2) or form a non overlapping but continuous chain(Example 1) as illustrated |
| * in examples below |
| * |
| * Example 1 |
| * A Storage has 2 Non Oevralpping but continuous Update Periods. |
| * MONTHLY with start time as now.month -13 months and end time as now.month -2months and |
| * DAILY with start time as now.day and end time as now.month -2months |
| * Then this Sorage will have an implied start time as now.month -13 month and end time as now.day |
| * |
| * Example 2 |
| * A Storage has 2 Overlapping Update Periods. |
| * MONTHLY with start time as now.month -13 months and end time as now.month -1months and |
| * DAILY with start time as now.day and end time as now.month -2months |
| * Then this Sorage will have an implied start time as now.month -13 month and end time as now.day |
| * |
| * @throws LensException |
| */ |
| void setStorageStartAndEndDate() throws LensException { |
| if (this.startTime != null && !this.isStorageTblsAtUpdatePeriodLevel) { |
| //If the times are already set and are not dependent of update period, no point setting times again. |
| return; |
| } |
| List<Date> startDates = new ArrayList<>(); |
| List<Date> endDates = new ArrayList<>(); |
| for (String storageTablePrefix : getValidStorageTableNames()) { |
| startDates.add(getCubeMetastoreClient().getStorageTableStartDate(storageTablePrefix, fact)); |
| endDates.add(getCubeMetastoreClient().getStorageTableEndDate(storageTablePrefix, fact)); |
| } |
| this.startTime = Collections.min(startDates); |
| this.endTime = Collections.max(endDates); |
| } |
| |
| private Set<String> getValidStorageTableNames() throws LensException { |
| if (!validUpdatePeriods.isEmpty()) { |
| // In this case skip invalid update periods and get storage tables only for valid ones. |
| Set<String> uniqueStorageTables = new HashSet<>(); |
| for (UpdatePeriod updatePeriod : validUpdatePeriods) { |
| uniqueStorageTables.add( |
| getCubeMetastoreClient().getStorageTableName(fact.getSourceFactName(), storageName, updatePeriod) |
| ); |
| } |
| return uniqueStorageTables; |
| } else { |
| //Get all storage tables. |
| return getCubeMetastoreClient().getStorageTables(fact, storageName); |
| } |
| } |
| |
| public void addAnswerableMeasurePhraseIndices(int index) { |
| answerableMeasurePhraseIndices.add(index); |
| } |
| |
| |
| @Override |
| public Candidate explode() throws LensException { |
| if (splitAtUpdatePeriodLevelIfReq().size() > 1) { |
| return new UnionCandidate(splitAtUpdatePeriodLevelIfReq(), getCubeQueryContext()); |
| } else { |
| return splitAtUpdatePeriodLevelIfReq().iterator().next(); |
| } |
| } |
| |
| @Override |
| public String getStorageString(String alias) { |
| return storageName + " " + alias; |
| } |
| |
| @Override |
| public AbstractCubeTable getBaseTable() { |
| return (AbstractCubeTable) cube; |
| } |
| |
| @Override |
| public StorageCandidate copy() throws LensException { |
| return new StorageCandidate(this); |
| } |
| |
| @Override |
| public boolean isPhraseAnswerable(QueriedPhraseContext phrase) throws LensException { |
| return phrase.isEvaluable(this); |
| } |
| |
| @Override |
| public AbstractCubeTable getTable() { |
| return (AbstractCubeTable) fact; |
| } |
| |
| public Optional<Date> getColumnStartTime(String column) { |
| Date startTime = null; |
| Map<String, String> propertiesMap = this.getFact().getSourceFactProperties(); |
| for (String key : propertiesMap.keySet()) { |
| if (key.contains(MetastoreConstants.FACT_COL_START_TIME_PFX)) { |
| String propCol = StringUtils.substringAfter(key, MetastoreConstants.FACT_COL_START_TIME_PFX); |
| if (column.equals(propCol)) { |
| startTime = MetastoreUtil.getDateFromProperty(propertiesMap.get(key), false, true); |
| } |
| } |
| } |
| return Optional.ofNullable(startTime); |
| } |
| |
| @Override |
| public Optional<Date> getColumnEndTime(String column) { |
| Date endTime = null; |
| Map<String, String> propertiesMap = this.getFact().getSourceFactProperties(); |
| for (String key : propertiesMap.keySet()) { |
| if (key.contains(MetastoreConstants.FACT_COL_END_TIME_PFX)) { |
| String propCol = StringUtils.substringAfter(key, MetastoreConstants.FACT_COL_END_TIME_PFX); |
| if (column.equals(propCol)) { |
| endTime = MetastoreUtil.getDateFromProperty(propertiesMap.get(key), false, true); |
| } |
| } |
| } |
| return Optional.ofNullable(endTime); |
| } |
| |
| @Override |
| public Collection<String> getColumns() { |
| if (factColumns == null) { |
| factColumns = fact.getValidColumns(); |
| if (factColumns == null) { |
| factColumns = fact.getAllFieldNames(); |
| } |
| } |
| return factColumns; |
| } |
| |
| @Override |
| public OptionalDouble getCost() { |
| return OptionalDouble.of(fact.weight()); |
| } |
| |
| @Override |
| public boolean contains(Candidate candidate) { |
| return this.equals(candidate); |
| } |
| |
| @Override |
| public Collection<Candidate> getChildren() { |
| return null; |
| } |
| |
| private void updatePartitionStorage(FactPartition part) throws LensException { |
| try { |
| if (getCubeMetastoreClient().factPartitionExists(fact, part, storageTable)) { |
| part.getStorageTables().add(storageTable); |
| part.setFound(true); |
| } |
| } catch (HiveException e) { |
| log.warn("Hive exception while getting storage table partition", e); |
| } |
| } |
| |
| /** |
| * Gets FactPartitions for the given fact using the following logic |
| * |
| * 1. Find the max update interval that will be used for the query. Lets assume time |
| * range is 15 Sep to 15 Dec and the fact has two storage with update periods as MONTHLY,DAILY,HOURLY. |
| * In this case the data for [15 sep - 1 oct)U[1 Dec - 15 Dec) will be answered by DAILY partitions |
| * and [1 oct - 1Dec) will be answered by MONTHLY partitions. The max interavl for this query will be MONTHLY. |
| * |
| * 2.Prune Storgaes that do not fall in the queries time range. |
| * {@link org.apache.lens.cube.metadata.CubeMetastoreClient#isStorageTableCandidateForRange(String, Date, Date)} |
| * |
| * 3. Iterate over max interavl . In out case it will give two months Oct and Nov. Find partitions for |
| * these two months.Check validity of FactPartitions for Oct and Nov |
| * via {@link #updatePartitionStorage(FactPartition)}. |
| * If the partition is missing, try getting partitions for the time range form other update periods (DAILY,HOURLY). |
| * This is achieved by calling getPartitions() recursively but passing only 2 update periods (DAILY,HOURLY) |
| * |
| * 4.If the monthly partitions are found, check for lookahead partitions and call getPartitions recursively for the |
| * remaining time intervals i.e, [15 sep - 1 oct) and [1 Dec - 15 Dec) |
| * |
| * TODO union : Move this into util. |
| */ |
| private boolean getPartitions(Date fromDate, Date toDate, String partCol, Set<FactPartition> partitions, |
| TreeSet<UpdatePeriod> updatePeriods, boolean addNonExistingParts, boolean failOnPartialData, |
| PartitionRangesForPartitionColumns missingPartitions) throws LensException { |
| if (fromDate.equals(toDate) || fromDate.after(toDate)) { |
| return true; |
| } |
| if (updatePeriods == null || updatePeriods.isEmpty()) { |
| return false; |
| } |
| |
| UpdatePeriod maxInterval = CubeFactTable.maxIntervalInRange(fromDate, toDate, updatePeriods); |
| if (maxInterval == null) { |
| log.info("No max interval for range: {} to {}", fromDate, toDate); |
| return false; |
| } |
| |
| if (maxInterval == UpdatePeriod.CONTINUOUS |
| && cubeQueryContext.getRangeWriter().getClass().equals(BetweenTimeRangeWriter.class)) { |
| FactPartition part = new FactPartition(partCol, fromDate, maxInterval, null, partWhereClauseFormat); |
| partitions.add(part); |
| part.getStorageTables().add(storageTable); |
| part = new FactPartition(partCol, toDate, maxInterval, null, partWhereClauseFormat); |
| partitions.add(part); |
| part.getStorageTables().add(storageTable); |
| this.participatingUpdatePeriods.add(maxInterval); |
| log.info("Added continuous fact partition for storage table {}", storageName); |
| return true; |
| } |
| |
| if (!getCubeMetastoreClient().partColExists(this.getFact(), storageName, partCol)) { |
| log.info("{} does not exist in {}", partCol, name); |
| return false; |
| } |
| |
| Date maxIntervalStorageTblStartDate = getStorageTableStartDate(maxInterval); |
| Date maxIntervalStorageTblEndDate = getStorageTableEndDate(maxInterval); |
| |
| TreeSet<UpdatePeriod> remainingIntervals = new TreeSet<>(updatePeriods); |
| remainingIntervals.remove(maxInterval); |
| if (!isCandidatePartiallyValidForTimeRange( |
| maxIntervalStorageTblStartDate, maxIntervalStorageTblEndDate, fromDate, toDate)) { |
| //Check the time range in remainingIntervals as maxInterval is not useful |
| return getPartitions(fromDate, toDate, partCol, partitions, remainingIntervals, |
| addNonExistingParts, failOnPartialData, missingPartitions); |
| } |
| |
| Date ceilFromDate = DateUtil.getCeilDate(fromDate.after(maxIntervalStorageTblStartDate) |
| ? fromDate : maxIntervalStorageTblStartDate, maxInterval); |
| Date floorToDate = DateUtil.getFloorDate(toDate.before(maxIntervalStorageTblEndDate) |
| ? toDate : maxIntervalStorageTblEndDate, maxInterval); |
| if (ceilFromDate.equals(floorToDate) || floorToDate.before(ceilFromDate)) { |
| return getPartitions(fromDate, toDate, partCol, partitions, remainingIntervals, |
| addNonExistingParts, failOnPartialData, missingPartitions); |
| } |
| |
| int lookAheadNumParts = getConf() |
| .getInt(CubeQueryConfUtil.getLookAheadPTPartsKey(maxInterval), CubeQueryConfUtil.DEFAULT_LOOK_AHEAD_PT_PARTS); |
| |
| int lookAheadNumTimeParts = getConf() |
| .getInt(CubeQueryConfUtil.getLookAheadTimePartsKey(maxInterval), CubeQueryConfUtil.DEFAULT_LOOK_AHEAD_TIME_PARTS); |
| |
| TimeRange.Iterable.Iterator iter = TimeRange.iterable(ceilFromDate, floorToDate, maxInterval, 1).iterator(); |
| // add partitions from ceilFrom to floorTo |
| while (iter.hasNext()) { |
| Date dt = iter.next(); |
| Date nextDt = iter.peekNext(); |
| FactPartition part = new FactPartition(partCol, dt, maxInterval, null, partWhereClauseFormat); |
| updatePartitionStorage(part); |
| log.debug("Storage tables containing Partition {} are: {}", part, part.getStorageTables()); |
| if (part.isFound()) { |
| log.debug("Adding existing partition {}", part); |
| partitions.add(part); |
| this.participatingUpdatePeriods.add(maxInterval); |
| log.debug("Looking for look ahead process time partitions for {}", part); |
| if (processTimePartCol == null) { |
| log.debug("processTimePartCol is null"); |
| } else if (partCol.equals(processTimePartCol)) { |
| log.debug("part column is process time col"); |
| } else if (updatePeriods.first().equals(maxInterval)) { |
| log.debug("Update period is the least update period"); |
| } else if ((iter.getNumIters() - iter.getCounter()) > lookAheadNumTimeParts) { |
| // see if this is the part of the last-n look ahead partitions |
| log.debug("Not a look ahead partition"); |
| } else { |
| log.debug("Looking for look ahead process time partitions for {}", part); |
| // check if finer partitions are required |
| // final partitions are required if no partitions from |
| // look-ahead |
| // process time are present |
| TimeRange.Iterable.Iterator processTimeIter = TimeRange.iterable(nextDt, lookAheadNumParts, maxInterval, 1) |
| .iterator(); |
| TimeRange.Iterable.Iterator timeIter = TimeRange.iterable(nextDt, lookAheadNumTimeParts, maxInterval, 1) |
| .iterator(); |
| while (processTimeIter.hasNext()) { |
| Date pdt = processTimeIter.next(); |
| Date nextPdt = processTimeIter.peekNext(); |
| FactPartition currFactPartition; |
| boolean allProcessTimePartitionsFound = true; |
| while (timeIter.hasNext()){ |
| Date date = timeIter.next(); |
| currFactPartition = new FactPartition(processTimePartCol, date, maxInterval, null, |
| partWhereClauseFormat); |
| updatePartitionStorage(currFactPartition); |
| if (!currFactPartition.isFound()) { |
| log.debug("Looked ahead process time partition {} is not found : " + currFactPartition); |
| allProcessTimePartitionsFound = false; |
| break; |
| } |
| } |
| if (allProcessTimePartitionsFound) { |
| log.debug("Finer parts not required for look-ahead partition :{}", part); |
| } else { |
| TreeSet<UpdatePeriod> newset = new TreeSet<UpdatePeriod>(); |
| newset.addAll(updatePeriods); |
| newset.remove(maxInterval); |
| log.debug("newset of update periods:{}", newset); |
| if (!newset.isEmpty()) { |
| // Get partitions for look ahead process time |
| log.debug("Looking for process time partitions between {} and {}", pdt, nextPdt); |
| Set<FactPartition> processTimeParts = getPartitions( |
| TimeRange.builder().fromDate(pdt).toDate(nextPdt).partitionColumn(processTimePartCol).build(), |
| newset, true, failOnPartialData, missingPartitions); |
| log.debug("Look ahead partitions: {}", processTimeParts); |
| TimeRange timeRange = TimeRange.builder().fromDate(dt).toDate(nextDt).build(); |
| for (FactPartition pPart : processTimeParts) { |
| log.debug("Looking for finer partitions in pPart: {}", pPart); |
| for (Date date : timeRange.iterable(pPart.getPeriod(), 1)) { |
| FactPartition innerPart = new FactPartition(partCol, date, pPart.getPeriod(), pPart, |
| partWhereClauseFormat); |
| updatePartitionStorage(innerPart); |
| innerPart.setFound(pPart.isFound()); |
| if (innerPart.isFound() || !failOnPartialData) { |
| this.participatingUpdatePeriods.add(maxInterval); |
| partitions.add(innerPart); |
| } |
| } |
| log.debug("added all sub partitions blindly in pPart: {}", pPart); |
| } |
| } |
| } |
| } |
| } |
| } else { |
| log.info("Partition:{} does not exist in any storage table", part); |
| if (!getPartitions(dt, nextDt, partCol, partitions, remainingIntervals, false, failOnPartialData, |
| missingPartitions)) { |
| log.debug("Adding non existing partition {}", part); |
| if (addNonExistingParts) { |
| // Add non existing partitions for all cases of whether we populate all non existing or not. |
| missingPartitions.add(part); |
| if (!failOnPartialData) { |
| this.participatingUpdatePeriods.add(maxInterval); |
| partitions.add(part); |
| part.getStorageTables().add(storageTable); |
| } |
| } else { |
| log.info("No finer granualar partitions exist for {}", part); |
| return false; |
| } |
| } else { |
| log.debug("Finer granualar partitions added for {}", part); |
| } |
| } |
| } |
| |
| return getPartitions(fromDate, ceilFromDate, partCol, partitions, remainingIntervals, |
| addNonExistingParts, failOnPartialData, missingPartitions) |
| && getPartitions(floorToDate, toDate, partCol, partitions, remainingIntervals, |
| addNonExistingParts, failOnPartialData, missingPartitions); |
| } |
| |
| private boolean isCandidatePartiallyValidForTimeRange(Date startDate, Date endDate, Date fromDate, Date toDate) { |
| return Stream.of(startDate, fromDate).max(naturalOrder()).orElse(startDate) |
| .before(Stream.of(endDate, toDate).min(naturalOrder()).orElse(endDate)); |
| } |
| |
| @Override |
| public boolean evaluateCompleteness(TimeRange timeRange, TimeRange queriedTimeRange, boolean failOnPartialData) |
| throws LensException { |
| // Check the measure tags. |
| if (!evaluateMeasuresCompleteness(timeRange)) { |
| log.info("Storage candidate:{} has partitions with incomplete data: {} for given ranges: {}", this, |
| dataCompletenessMap, cubeQueryContext.getTimeRanges()); |
| if (failOnPartialData) { |
| return false; |
| } |
| } |
| PartitionRangesForPartitionColumns missingParts = new PartitionRangesForPartitionColumns(); |
| PruneCauses<Candidate> storagePruningMsgs = cubeQueryContext.getStoragePruningMsgs(); |
| Set<String> unsupportedTimeDims = Sets.newHashSet(); |
| Set<String> partColsQueried = Sets.newHashSet(); |
| partColsQueried.add(timeRange.getPartitionColumn()); |
| StringBuilder extraWhereClauseFallback = new StringBuilder(); |
| Set<FactPartition> rangeParts = getPartitions(timeRange, validUpdatePeriods, true, failOnPartialData, missingParts); |
| String partCol = timeRange.getPartitionColumn(); |
| boolean partColNotSupported = rangeParts.isEmpty(); |
| |
| if (storagePruningMsgs.containsKey(this)) { |
| List<CandidateTablePruneCause> causes = storagePruningMsgs.get(this); |
| // Find the PART_COL_DOES_NOT_EXISTS |
| for (CandidateTablePruneCause cause : causes) { |
| if (cause.getCause().equals(CandidateTablePruneCode.PART_COL_DOES_NOT_EXIST)) { |
| partColNotSupported &= cause.getNonExistantPartCols().contains(partCol); |
| } |
| } |
| } else { |
| partColNotSupported = false; |
| } |
| TimeRange prevRange = timeRange; |
| String sep = ""; |
| while (rangeParts.isEmpty()) { |
| String timeDim = cubeQueryContext.getBaseCube().getTimeDimOfPartitionColumn(partCol); |
| if (getFact() instanceof CubeFactTable) { |
| if (partColNotSupported && !((CubeFactTable) getFact()).hasColumn(timeDim)) { |
| unsupportedTimeDims.add( |
| cubeQueryContext.getBaseCube().getTimeDimOfPartitionColumn(timeRange.getPartitionColumn()) |
| ); |
| break; |
| } |
| } |
| TimeRange fallBackRange = getFallbackRange(prevRange, this.getFact().getSourceFactName(), cubeQueryContext); |
| log.info("No partitions for range:{}. fallback range: {}", timeRange, fallBackRange); |
| if (fallBackRange == null) { |
| break; |
| } |
| partColsQueried.add(fallBackRange.getPartitionColumn()); |
| rangeParts = getPartitions(fallBackRange, validUpdatePeriods, true, failOnPartialData, missingParts); |
| extraWhereClauseFallback.append(sep).append( |
| prevRange.toTimeDimWhereClause(cubeQueryContext.getAliasForTableName(cubeQueryContext.getCube()), timeDim) |
| ); |
| sep = " AND "; |
| prevRange = fallBackRange; |
| partCol = prevRange.getPartitionColumn(); |
| if (!rangeParts.isEmpty()) { |
| break; |
| } |
| } |
| // Add all the partitions. participatingPartitions contains all the partitions for previous time ranges also. |
| rangeToPartitions.put(queriedTimeRange, rangeParts); |
| numQueriedParts += rangeParts.size(); |
| if (!unsupportedTimeDims.isEmpty()) { |
| log.info("Not considering storage candidate:{} as it doesn't support time dimensions: {}", this, |
| unsupportedTimeDims); |
| cubeQueryContext.addStoragePruningMsg(this, timeDimNotSupported(unsupportedTimeDims)); |
| return false; |
| } |
| Set<String> nonExistingParts = missingParts.toSet(partColsQueried); |
| // TODO union : Relook at this. |
| nonExistingPartitions.addAll(nonExistingParts); |
| if (rangeParts.size() == 0 || (failOnPartialData && !nonExistingParts.isEmpty())) { |
| log.info("Not considering storage candidate:{} as no partitions for fallback range:{}", this, timeRange); |
| return false; |
| } |
| String extraWhere = extraWhereClauseFallback.toString(); |
| if (!StringUtils.isEmpty(extraWhere)) { |
| rangeToExtraWhereFallBack.put(queriedTimeRange, extraWhere); |
| } |
| return true; |
| } |
| |
| @Override |
| public Set<FactPartition> getParticipatingPartitions() { |
| Set<FactPartition> allPartitions = new HashSet<>(numQueriedParts); |
| for (Set<FactPartition> rangePartitions : rangeToPartitions.values()) { |
| allPartitions.addAll(rangePartitions); |
| } |
| return allPartitions; |
| } |
| |
| private boolean evaluateMeasuresCompleteness(TimeRange timeRange) throws LensException { |
| if (getCubeMetastoreClient() == null || !getCubeMetastoreClient().isDataCompletenessCheckEnabled()) { |
| log.info("Skipping availability check for the fact table: {} as dataCompleteness check is not enabled", fact); |
| return true; |
| } |
| String factDataCompletenessTag = fact.getDataCompletenessTag(); |
| if (factDataCompletenessTag == null) { |
| log.info("Not checking completeness for the fact table:{} as the dataCompletenessTag is not set", fact); |
| return true; |
| } |
| Set<String> measureTag = new HashSet<>(); |
| Map<String, String> tagToMeasureOrExprMap = new HashMap<>(); |
| |
| processExpressionsForCompleteness(cubeQueryContext, measureTag, tagToMeasureOrExprMap); |
| |
| Set<String> measures = cubeQueryContext.getQueriedMsrs(); |
| if (measures == null) { |
| measures = new HashSet<>(); |
| } |
| for (String measure : measures) { |
| processCubeColForDataCompleteness(cubeQueryContext, measure, measure, measureTag, tagToMeasureOrExprMap); |
| } |
| //Checking if dataCompletenessTag is set for the fact |
| if (measureTag.isEmpty()) { |
| log.info("No Queried measures with the dataCompletenessTag, hence skipping the availability check"); |
| return true; |
| } |
| // default completenessTag will be true |
| boolean isDataComplete = true; |
| DataCompletenessChecker completenessChecker = getCubeMetastoreClient().getCompletenessChecker(); |
| DateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); |
| formatter.setTimeZone(TimeZone.getTimeZone("UTC")); |
| if (!timeRange.getPartitionColumn().equals(completenessPartCol)) { |
| log.info("Completeness check not available for partCol:{}", timeRange.getPartitionColumn()); |
| return true; |
| } |
| Date from = timeRange.getFromDate(); |
| Date to = timeRange.getToDate(); |
| Map<String, Map<Date, Float>> completenessMap = completenessChecker |
| .getCompleteness(factDataCompletenessTag, from, to, measureTag); |
| if (completenessMap != null && !completenessMap.isEmpty()) { |
| for (Map.Entry<String, Map<Date, Float>> measureCompleteness : completenessMap.entrySet()) { |
| String tag = measureCompleteness.getKey(); |
| for (Map.Entry<Date, Float> completenessResult : measureCompleteness.getValue().entrySet()) { |
| if (completenessResult.getValue() < completenessThreshold) { |
| log.info("Completeness for the measure_tag {} is {}, threshold: {}, for the hour {}", tag, |
| completenessResult.getValue(), completenessThreshold, formatter.format(completenessResult.getKey())); |
| String measureorExprFromTag = tagToMeasureOrExprMap.get(tag); |
| dataCompletenessMap.computeIfAbsent(measureorExprFromTag, k -> new HashMap<>()) |
| .put(formatter.format(completenessResult.getKey()), completenessResult.getValue()); |
| // set completeness to false if availability for measure is below threshold |
| isDataComplete = false; |
| } |
| } |
| } |
| } |
| return isDataComplete; |
| } |
| |
| private Set<FactPartition> getPartitions(TimeRange timeRange, TreeSet<UpdatePeriod> updatePeriods, |
| boolean addNonExistingParts, boolean failOnPartialData, PartitionRangesForPartitionColumns missingParts) |
| throws LensException { |
| Set<FactPartition> partitions = new TreeSet<>(); |
| if (timeRange != null && timeRange.isCoverableBy(updatePeriods)) { |
| getPartitions(timeRange.getFromDate(), timeRange.getToDate(), timeRange.getPartitionColumn(), |
| partitions, updatePeriods, addNonExistingParts, failOnPartialData, missingParts); |
| } |
| return partitions; |
| } |
| |
| @Override |
| public boolean isExpressionEvaluable(ExpressionResolver.ExpressionContext expr) { |
| return expr.isEvaluable(this); |
| } |
| |
| @Override |
| public boolean isExpressionEvaluable(String expr) { |
| return isExpressionEvaluable( |
| getCubeQueryContext().getExprCtx().getExpressionContext( |
| expr, getCubeQueryContext().getAliasForTableName(getBaseTable().getName())) |
| ); |
| } |
| |
| @Override |
| public boolean isDimAttributeEvaluable(String dim) throws LensException { |
| return getCubeQueryContext().getDeNormCtx() |
| .addRefUsage(getCubeQueryContext(), this, dim, getCubeQueryContext().getCube().getName()); |
| } |
| @Override |
| public boolean equals(Object obj) { |
| if (super.equals(obj)) { |
| return true; |
| } |
| |
| if (obj == null || !(obj instanceof StorageCandidate)) { |
| return false; |
| } |
| |
| StorageCandidate storageCandidateObj = (StorageCandidate) obj; |
| //Assuming that same instance of cube and fact will be used across StorageCandidate s and hence relying directly |
| //on == check for these. |
| return (this.cube == storageCandidateObj.cube && this.fact == storageCandidateObj.fact && this.storageTable |
| .equals(storageCandidateObj.storageTable)); |
| } |
| |
| @Override |
| public int hashCode() { |
| return this.storageTable.hashCode(); |
| } |
| |
| @Override |
| public String toString() { |
| return getResolvedName(); |
| } |
| |
| void addValidUpdatePeriod(UpdatePeriod updatePeriod) { |
| this.validUpdatePeriods.add(updatePeriod); |
| } |
| |
| public String getAliasForTable(String alias) { |
| String database = SessionState.get().getCurrentDatabase(); |
| String ret; |
| if (alias == null || alias.isEmpty()) { |
| ret = getResolvedName(); |
| } else { |
| ret = getResolvedName() + " " + alias; |
| } |
| if (StringUtils.isNotBlank(database) && !"default".equalsIgnoreCase(database)) { |
| ret = database + "." + ret; |
| } |
| return ret; |
| } |
| |
| boolean isUpdatePeriodUseful(UpdatePeriod updatePeriod) { |
| return getCubeQueryContext().getTimeRanges().stream() |
| .anyMatch(timeRange -> isUpdatePeriodUseful(timeRange, updatePeriod)); |
| } |
| |
| /** |
| * Is the update period useful for this time range. e.g. for a time range of hours and days, monthly |
| * and yearly update periods are useless. DAILY and HOURLY are useful. It further checks if the update |
| * period answers the range at least partially based on start and end times configured at update period |
| * level or at storage or fact level. |
| * @param timeRange The time range |
| * @param updatePeriod Update period |
| * @return Whether it's useless |
| */ |
| private boolean isUpdatePeriodUseful(TimeRange timeRange, UpdatePeriod updatePeriod) { |
| try { |
| if (!timeRange.truncate(getStorageTableStartDate(updatePeriod), |
| getStorageTableEndDate(updatePeriod)).isValid()) { |
| return false; |
| } |
| Date storageTblStartDate = getStorageTableStartDate(updatePeriod); |
| Date storageTblEndDate = getStorageTableEndDate(updatePeriod); |
| TimeRange.builder() //TODO date calculation to move to util method and resued |
| .fromDate(timeRange.getFromDate().after(storageTblStartDate) ? timeRange.getFromDate() : storageTblStartDate) |
| .toDate(timeRange.getToDate().before(storageTblEndDate) ? timeRange.getToDate() : storageTblEndDate) |
| .partitionColumn(timeRange.getPartitionColumn()) |
| .build() |
| .truncate(updatePeriod); |
| return true; |
| } catch (LensException e) { |
| return false; |
| } |
| } |
| |
| /** |
| * Is time range coverable based on valid update periods of this storage candidate |
| * |
| * @param timeRange |
| * @return |
| * @throws LensException |
| */ |
| public boolean isTimeRangeCoverable(TimeRange timeRange) throws LensException { |
| return isTimeRangeCoverable(timeRange.getFromDate(), timeRange.getToDate(), validUpdatePeriods); |
| } |
| |
| /* |
| * Is the time range coverable by given update periods. |
| * Extracts the max update period, then extracts maximum amount of range from the middle that this update |
| * period can cover. Then recurses on the remaining ranges on the left and right side of the extracted chunk |
| * using one less update period. |
| * |
| * @param timeRangeStart |
| * @param timeRangeEnd |
| * @param intervals Update periods to check |
| * @return Whether time range is coverable by provided update periods or not. |
| */ |
| private boolean isTimeRangeCoverable(Date timeRangeStart, Date timeRangeEnd, |
| Set<UpdatePeriod> intervals) throws LensException { |
| if (timeRangeStart.equals(timeRangeEnd) || timeRangeStart.after(timeRangeEnd)) { |
| return true; |
| } |
| if (intervals == null || intervals.isEmpty()) { |
| return false; |
| } |
| |
| UpdatePeriod maxInterval = CubeFactTable.maxIntervalInRange(timeRangeStart, timeRangeEnd, intervals); |
| if (maxInterval == null) { |
| return false; |
| } |
| |
| if (maxInterval == UpdatePeriod.CONTINUOUS |
| && getCubeQueryContext().getRangeWriter().getClass().equals(BetweenTimeRangeWriter.class)) { |
| return true; |
| } |
| |
| Date maxIntervalStorageTableStartDate = getStorageTableStartDate(maxInterval); |
| Date maxIntervalStorageTableEndDate = getStorageTableEndDate(maxInterval); |
| Set<UpdatePeriod> remainingIntervals = Sets.difference(intervals, Sets.newHashSet(maxInterval)); |
| |
| if (!isCandidatePartiallyValidForTimeRange( |
| maxIntervalStorageTableStartDate, maxIntervalStorageTableEndDate, timeRangeStart, timeRangeEnd)) { |
| //Check the time range in remainingIntervals as maxInterval is not useful |
| return isTimeRangeCoverable(timeRangeStart, timeRangeEnd, remainingIntervals); |
| } |
| |
| Date ceilFromDate = DateUtil.getCeilDate(timeRangeStart.after(maxIntervalStorageTableStartDate) |
| ? timeRangeStart : maxIntervalStorageTableStartDate, maxInterval); |
| Date floorToDate = DateUtil.getFloorDate(timeRangeEnd.before(maxIntervalStorageTableEndDate) |
| ? timeRangeEnd : maxIntervalStorageTableEndDate, maxInterval); |
| if (ceilFromDate.equals(floorToDate) || floorToDate.before(ceilFromDate)) { |
| return isTimeRangeCoverable(timeRangeStart, timeRangeEnd, remainingIntervals); |
| } |
| |
| //ceilFromDate to floorToDate time range is covered by maxInterval (though there may be holes.. but that's ok) |
| //Check the remaining part of time range in remainingIntervals |
| return isTimeRangeCoverable(timeRangeStart, ceilFromDate, remainingIntervals) |
| && isTimeRangeCoverable(floorToDate, timeRangeEnd, remainingIntervals); |
| } |
| |
| private Date getStorageTableStartDate(UpdatePeriod interval) throws LensException { |
| if (!isStorageTblsAtUpdatePeriodLevel) { |
| //In this case the start time and end time is at Storage Level and will be same for all update periods. |
| return this.startTime; |
| } |
| return getCubeMetastoreClient().getStorageTableStartDate( |
| getCubeMetastoreClient().getStorageTableName(fact.getSourceFactName(), storageName, interval), fact); |
| } |
| |
| private Date getStorageTableEndDate(UpdatePeriod interval) throws LensException { |
| if (!isStorageTblsAtUpdatePeriodLevel) { |
| //In this case the start time and end time is at Storage Level and will be same for all update periods. |
| return this.endTime; |
| } |
| return getCubeMetastoreClient().getStorageTableEndDate( |
| getCubeMetastoreClient().getStorageTableName(fact.getSourceFactName(), storageName, interval), fact); |
| } |
| |
| |
| public String getResolvedName() { |
| if (resolvedName == null) { |
| return storageTable; |
| } |
| return resolvedName; |
| } |
| |
| /** |
| * Splits the Storage Candidates into multiple Storage Candidates if storage candidate has multiple |
| * storage tables (one per update period) |
| * |
| * @return |
| * @throws LensException |
| */ |
| public Collection<StorageCandidate> splitAtUpdatePeriodLevelIfReq() throws LensException { |
| if (!isStorageTblsAtUpdatePeriodLevel) { |
| return Lists.newArrayList(this); // No need to explode in this case |
| } |
| return getPeriodSpecificStorageCandidates(); |
| } |
| |
| private Collection<StorageCandidate> getPeriodSpecificStorageCandidates() throws LensException { |
| if (periodSpecificStorageCandidates == null) { |
| List<StorageCandidate> periodSpecificScList = new ArrayList<>(participatingUpdatePeriods.size()); |
| StorageCandidate updatePeriodSpecificSc; |
| for (UpdatePeriod period : participatingUpdatePeriods) { |
| updatePeriodSpecificSc = copy(); |
| updatePeriodSpecificSc.setResolvedName(getCubeMetastoreClient().getStorageTableName(fact.getSourceFactName(), |
| storageName, period)); |
| updatePeriodSpecificSc.isStorageTblsAtUpdatePeriodLevel = false; |
| updatePeriodSpecificSc.truncatePartitions(period); |
| periodSpecificScList.add(updatePeriodSpecificSc); |
| } |
| periodSpecificStorageCandidates = periodSpecificScList; |
| } |
| return periodSpecificStorageCandidates; |
| } |
| |
| /** |
| * Truncates partitions in {@link #rangeToPartitions} such that only partitions belonging to |
| * the passed undatePeriod are retained. |
| * @param updatePeriod |
| */ |
| private void truncatePartitions(UpdatePeriod updatePeriod) { |
| Iterator<Map.Entry<TimeRange, Set<FactPartition>>> rangeItr = rangeToPartitions.entrySet().iterator(); |
| while (rangeItr.hasNext()) { |
| Map.Entry<TimeRange, Set<FactPartition>> rangeEntry = rangeItr.next(); |
| rangeEntry.getValue().removeIf(factPartition -> !factPartition.getPeriod().equals(updatePeriod)); |
| rangeEntry.getValue().forEach(factPartition -> { |
| factPartition.getStorageTables().remove(storageTable); |
| factPartition.getStorageTables().add(resolvedName); |
| }); |
| if (rangeEntry.getValue().isEmpty()) { |
| rangeItr.remove(); |
| } |
| } |
| } |
| |
| @Override |
| public StorageCandidateHQLContext toQueryWriterContext(Map<Dimension, CandidateDim> dimsToQuery, |
| CubeQueryContext rootCubeQueryContext) throws LensException { |
| DefaultQueryAST ast = DefaultQueryAST.fromStorageCandidate(null, getCubeQueryContext()); |
| ast.copyFrom(getCubeQueryContext()); |
| return new StorageCandidateHQLContext(this, Maps.newHashMap(dimsToQuery), ast, rootCubeQueryContext); |
| } |
| |
| @Override |
| public Set<Integer> decideMeasurePhrasesToAnswer(Set<Integer> measureIndices) { |
| answerableMeasurePhraseIndices.retainAll(measureIndices); |
| return answerableMeasurePhraseIndices; |
| } |
| } |
