11970

Build # 11970

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/24ac98f1dd944ffb11e4425f8a56d50b703f1543">24ac98f1d<a href="https://github.com/apache/carbondata/commit/24ac98f1dd944ffb11e4425f8a56d50b703f1543">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/a3768015fa62504f1abd9b1ca09f13c04a38943e">a3768015f</a>">a3768015f</a><a href="https://github.com/apache/carbondata/commit/24ac98f1dd944ffb11e4425f8a56d50b703f1543"> into e7c96abef">e7c96abef</a>

Run Details

13 of 13 new or added lines in 1 file covered. (100.0%)

64950 of 83493 relevant lines covered (77.79%)

1.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

4.57

/core/src/main/java/org/apache/carbondata/core/datamap/DataMapProvider.java

/*
 * 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.carbondata.core.datamap;

import java.io.IOException;
import java.util.*;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
import org.apache.carbondata.common.exceptions.sql.NoSuchDataMapException;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.dev.DataMapFactory;
import org.apache.carbondata.core.datamap.status.DataMapSegmentStatusUtil;
import org.apache.carbondata.core.datamap.status.DataMapStatusManager;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatus;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import com.google.gson.Gson;
import org.apache.log4j.Logger;

/**
 * DataMap is a accelerator for certain type of query. Developer can add new DataMap
 * implementation to improve query performance.
 *
 * Currently two types of DataMap are supported
 * <ol>
 *   <li> MVDataMap: materialized view type of DataMap to accelerate olap style query,
 * like SPJG query (select, predicate, join, groupby) </li>
 *   <li> DataMap: index type of DataMap to accelerate filter query </li>
 * </ol>
 *
 * <p>
 * In following command <br>
 * {@code CREATE DATAMAP dm ON TABLE main USING 'provider'}, <br>
 * the <b>provider</b> string can be a short name or class name of the DataMap implementation.
 *
 * <br>Currently CarbonData supports following provider:
 * <ol>
 *   <li> preaggregate: pre-aggregate table of single table </li>
 *   <li> timeseries: pre-aggregate table based on time dimension of the table </li>
 *   <li> lucene: index backed by Apache Lucene </li>
 *   <li> bloomfilter: index backed by Bloom Filter </li>
 * </ol>
 *
 * @since 1.4.0
 */
@InterfaceAudience.Internal
public abstract class DataMapProvider {

  private CarbonTable mainTable;
  private DataMapSchema dataMapSchema;

  private Logger LOGGER = LogServiceFactory.getLogService(this.getClass().getCanonicalName());

  public DataMapProvider(CarbonTable mainTable, DataMapSchema dataMapSchema) {
    this.mainTable = mainTable;
    this.dataMapSchema = dataMapSchema;
  }

  protected final CarbonTable getMainTable() {
    return mainTable;
  }

  public final DataMapSchema getDataMapSchema() {
    return dataMapSchema;
  }

  /**
   * Initialize a datamap's metadata.
   * This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
   * Implementation should initialize metadata for datamap, like creating table
   */
  public abstract void initMeta(String ctasSqlStatement) throws MalformedDataMapCommandException,
      IOException;

  /**
   * Initialize a datamap's data.
   * This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
   * Implementation should initialize data for datamap, like creating data folders
   */
  public void initData() { }

  /**
   * Opposite operation of {@link #initMeta(String)}.
   * This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
   * Implementation should clean all meta for the datamap
   */
  public abstract void cleanMeta() throws IOException;

  /**
   * Opposite operation of {@link #initData()}.
   * This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
   * Implementation should clean all data for the datamap
   */
  public abstract void cleanData();

  /**
   * Rebuild the datamap by loading all existing data from mainTable
   * This is called when refreshing the datamap when
   * 1. after datamap creation and if `autoRefreshDataMap` is set to true
   * 2. user manually trigger REBUILD DATAMAP command
   */
  public boolean rebuild() throws IOException, NoSuchDataMapException {
    if (null == dataMapSchema.getRelationIdentifier()) {
      return false;
    }
    String newLoadName = "";
    String segmentMap = "";
    AbsoluteTableIdentifier dataMapTableAbsoluteTableIdentifier = AbsoluteTableIdentifier
        .from(dataMapSchema.getRelationIdentifier().getTablePath(),
            dataMapSchema.getRelationIdentifier().getDatabaseName(),
            dataMapSchema.getRelationIdentifier().getTableName());
    SegmentStatusManager segmentStatusManager =
        new SegmentStatusManager(dataMapTableAbsoluteTableIdentifier);
    Map<String, List<String>> segmentMapping = new HashMap<>();
    // Acquire table status lock to handle concurrent dataloading
    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
    try {
      if (carbonLock.lockWithRetries()) {
        LOGGER.info(
            "Acquired lock for table" + dataMapSchema.getRelationIdentifier().getDatabaseName()
                + "." + dataMapSchema.getRelationIdentifier().getTableName()
                + " for table status updation");
        String dataMapTableMetadataPath =
            CarbonTablePath.getMetadataPath(dataMapSchema.getRelationIdentifier().getTablePath());
        LoadMetadataDetails[] loadMetaDataDetails =
            SegmentStatusManager.readLoadMetadata(dataMapTableMetadataPath);
        List<LoadMetadataDetails> listOfLoadFolderDetails =
            new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
        Collections.addAll(listOfLoadFolderDetails, loadMetaDataDetails);
        if (dataMapSchema.isLazy()) {
          // check if rebuild to datamap is already in progress and throw exception
          if (!listOfLoadFolderDetails.isEmpty()) {
            for (LoadMetadataDetails loadMetaDetail : loadMetaDataDetails) {
              if ((loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS
                  || loadMetaDetail.getSegmentStatus()
                  == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) && SegmentStatusManager
                  .isLoadInProgress(dataMapTableAbsoluteTableIdentifier,
                      loadMetaDetail.getLoadName())) {
                throw new RuntimeException("Rebuild to datamap " + dataMapSchema.getDataMapName()
                    + " is already in progress");
              }
            }
          }
        }
        boolean isFullRefresh = false;
        if (null != dataMapSchema.getProperties().get("full_refresh")) {
          isFullRefresh = Boolean.valueOf(dataMapSchema.getProperties().get("full_refresh"));
        }
        if (!isFullRefresh) {
          if (!getSpecificSegmentsTobeLoaded(segmentMapping, listOfLoadFolderDetails)) {
            return false;
          }
          segmentMap = new Gson().toJson(segmentMapping);
        } else {
          List<RelationIdentifier> relationIdentifiers = dataMapSchema.getParentTables();
          for (RelationIdentifier relationIdentifier : relationIdentifiers) {
            List<String> mainTableSegmentList =
                DataMapUtil.getMainTableValidSegmentList(relationIdentifier);
            if (mainTableSegmentList.isEmpty()) {
              return false;
            }
            segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                + relationIdentifier.getTableName(), mainTableSegmentList);
          }
          segmentMap = new Gson().toJson(segmentMapping);
        }

        // To handle concurrent dataloading to datamap, create new loadMetaEntry and
        // set segmentMap to new loadMetaEntry and pass new segmentId with load command
        LoadMetadataDetails loadMetadataDetail = new LoadMetadataDetails();
        String segmentId =
            String.valueOf(SegmentStatusManager.createNewSegmentId(loadMetaDataDetails));
        loadMetadataDetail.setLoadName(segmentId);
        loadMetadataDetail.setSegmentStatus(SegmentStatus.INSERT_IN_PROGRESS);
        loadMetadataDetail.setExtraInfo(segmentMap);
        listOfLoadFolderDetails.add(loadMetadataDetail);
        newLoadName = segmentId;

        SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath
                .getTableStatusFilePath(dataMapSchema.getRelationIdentifier().getTablePath()),
            listOfLoadFolderDetails
                .toArray(new LoadMetadataDetails[listOfLoadFolderDetails.size()]));
      } else {
        LOGGER.error(
            "Not able to acquire the lock for Table status updation for table " + dataMapSchema
                .getRelationIdentifier().getDatabaseName() + "." + dataMapSchema
                .getRelationIdentifier().getTableName());
        DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName());
        return false;
      }
    } finally {
      if (carbonLock.unlock()) {
        LOGGER.info("Table unlocked successfully after table status updation" + dataMapSchema
            .getRelationIdentifier().getDatabaseName() + "." + dataMapSchema.getRelationIdentifier()
            .getTableName());
      } else {
        LOGGER.error("Unable to unlock Table lock for table" + dataMapSchema.getRelationIdentifier()
            .getDatabaseName() + "." + dataMapSchema.getRelationIdentifier().getTableName()
            + " during table status updation");
      }
    }
    return rebuildInternal(newLoadName, segmentMapping);
  }

  /**
   * This method will compare mainTable and dataMapTable segment List and loads only newly added
   * segment from main table to dataMap table.
   * In case if mainTable is compacted, then based on dataMap to mainTables segmentMapping, dataMap
   * will be loaded
   * Eg:
   * case 1: Consider mainTableSegmentList: {0, 1, 2}, dataMapToMainTable segmentMap:
   * { 0 -> 0, 1-> 1,2}. If (1, 2) segments of main table are compacted to 1.1 and new segment (3)
   * is loaded to main table, then mainTableSegmentList will be updated to{0, 1.1, 3}.
   * In this case, segment (1) of dataMap table will be marked for delete, and new segment
   * {2 -> 1.1, 3} will be loaded to dataMap table
   * case 2: Consider mainTableSegmentList: {0, 1, 2, 3}, dataMapToMainTable segmentMap:
   * { 0 -> 0,1,2, 1-> 3}. If (1, 2) segments of main table are compacted to 1.1 and new segment
   * (4) is loaded to main table, then mainTableSegmentList will be updated to {0, 1.1, 3, 4}.
   * In this case, segment (0) of dataMap table will be marked for delete and segment (0) of
   * main table will be added to validSegmentList which needs to be loaded again. Now, new dataMap
   * table segment (2) with main table segmentList{2 -> 1.1, 4, 0} will be loaded to dataMap table.
   * dataMapToMainTable segmentMap will be updated to {1 -> 3, 2 -> 1.1, 4, 0} after rebuild
   */
  private boolean getSpecificSegmentsTobeLoaded(Map<String, List<String>> segmentMapping,
      List<LoadMetadataDetails> listOfLoadFolderDetails) throws IOException {
    List<RelationIdentifier> relationIdentifiers = dataMapSchema.getParentTables();
    // invalidDataMapSegmentList holds segment list which needs to be marked for delete
    HashSet<String> invalidDataMapSegmentList = new HashSet<>();
    if (listOfLoadFolderDetails.isEmpty()) {
      // If segment Map is empty, load all valid segments from main tables to dataMap
      for (RelationIdentifier relationIdentifier : relationIdentifiers) {
        List<String> mainTableSegmentList =
            DataMapUtil.getMainTableValidSegmentList(relationIdentifier);
        // If mainTableSegmentList is empty, no need to trigger load command
        // TODO: handle in case of multiple tables load to datamap table
        if (mainTableSegmentList.isEmpty()) {
          return false;
        }
        segmentMapping.put(
            relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT + relationIdentifier
                .getTableName(), mainTableSegmentList);
      }
    } else {
      for (RelationIdentifier relationIdentifier : relationIdentifiers) {
        List<String> dataMapTableSegmentList = new ArrayList<>();
        // Get all segments for parent relationIdentifier
        List<String> mainTableSegmentList =
            DataMapUtil.getMainTableValidSegmentList(relationIdentifier);
        boolean ifTableStatusUpdateRequired = false;
        for (LoadMetadataDetails loadMetaDetail : listOfLoadFolderDetails) {
          if (loadMetaDetail.getSegmentStatus() == SegmentStatus.SUCCESS
              || loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS) {
            Map<String, List<String>> segmentMaps =
                DataMapSegmentStatusUtil.getSegmentMap(loadMetaDetail.getExtraInfo());
            String mainTableMetaDataPath =
                CarbonTablePath.getMetadataPath(relationIdentifier.getTablePath());
            LoadMetadataDetails[] parentTableLoadMetaDataDetails =
                SegmentStatusManager.readLoadMetadata(mainTableMetaDataPath);
            String table = relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                + relationIdentifier.getTableName();
            for (String segmentId : mainTableSegmentList) {
              // In case if dataMap segment(0) is mapped to mainTable segments{0,1,2} and if
              // {0,1,2} segments of mainTable are compacted to 0.1. Then,
              // on next rebuild/load to dataMap, no need to load segment(0.1) again. Update the
              // segmentMapping of dataMap segment from {0,1,2} to {0.1}
              if (!checkIfSegmentsToBeReloaded(parentTableLoadMetaDataDetails,
                  segmentMaps.get(table), segmentId)) {
                ifTableStatusUpdateRequired = true;
                // Update loadMetaDetail with updated segment info and clear old segmentMap
                Map<String, List<String>> updatedSegmentMap = new HashMap<>();
                List<String> segmentList = new ArrayList<>();
                segmentList.add(segmentId);
                updatedSegmentMap.put(table, segmentList);
                dataMapTableSegmentList.add(segmentId);
                loadMetaDetail.setExtraInfo(new Gson().toJson(updatedSegmentMap));
                segmentMaps.get(table).clear();
              }
            }
            dataMapTableSegmentList.addAll(segmentMaps.get(table));
          }
        }
        List<String> dataMapSegmentList = new ArrayList<>(dataMapTableSegmentList);
        dataMapTableSegmentList.removeAll(mainTableSegmentList);
        mainTableSegmentList.removeAll(dataMapSegmentList);
        if (ifTableStatusUpdateRequired && mainTableSegmentList.isEmpty()) {
          SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath
                  .getTableStatusFilePath(dataMapSchema.getRelationIdentifier().getTablePath()),
              listOfLoadFolderDetails
                  .toArray(new LoadMetadataDetails[listOfLoadFolderDetails.size()]));
          return false;
        } else if (mainTableSegmentList.isEmpty()) {
          return false;
        }
        if (!dataMapTableSegmentList.isEmpty()) {
          List<String> invalidMainTableSegmentList = new ArrayList<>();
          // validMainTableSegmentList holds segment list which needs to be loaded again
          HashSet<String> validMainTableSegmentList = new HashSet<>();

          // For dataMap segments which are not in main table segment list(if main table
          // is compacted), iterate over those segments and get dataMap segments which needs to
          // be marked for delete and main table segments which needs to be loaded again
          for (String segmentId : dataMapTableSegmentList) {
            for (LoadMetadataDetails loadMetaDetail : listOfLoadFolderDetails) {
              if (loadMetaDetail.getSegmentStatus() == SegmentStatus.SUCCESS
                  || loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS) {
                Map<String, List<String>> segmentMaps =
                    DataMapSegmentStatusUtil.getSegmentMap(loadMetaDetail.getExtraInfo());
                List<String> segmentIds = segmentMaps.get(
                    relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                        + relationIdentifier.getTableName());
                if (segmentIds.contains(segmentId)) {
                  segmentIds.remove(segmentId);
                  validMainTableSegmentList.addAll(segmentIds);
                  invalidMainTableSegmentList.add(segmentId);
                  invalidDataMapSegmentList.add(loadMetaDetail.getLoadName());
                }
              }
            }
          }
          // remove invalid segment from validMainTableSegmentList if present
          validMainTableSegmentList.removeAll(invalidMainTableSegmentList);
          // Add all valid segments of main table which needs to be loaded again
          mainTableSegmentList.addAll(validMainTableSegmentList);
          segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
              + relationIdentifier.getTableName(), mainTableSegmentList);
        } else {
          segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
              + relationIdentifier.getTableName(), mainTableSegmentList);
        }
      }
    }
    // Remove invalid datamap segments
    if (!invalidDataMapSegmentList.isEmpty()) {
      for (LoadMetadataDetails loadMetadataDetail : listOfLoadFolderDetails) {
        if (invalidDataMapSegmentList.contains(loadMetadataDetail.getLoadName())) {
          loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
        }
      }
    }
    return true;
  }

  /**
   * This method checks if dataMap table segment has to be reloaded again or not
   */
  private boolean checkIfSegmentsToBeReloaded(LoadMetadataDetails[] loadMetaDataDetails,
      List<String> segmentIds, String segmentId) {
    boolean isToBeLoadedAgain = true;
    List<String> mergedSegments = new ArrayList<>();
    for (LoadMetadataDetails loadMetadataDetail : loadMetaDataDetails) {
      if (null != loadMetadataDetail.getMergedLoadName() && loadMetadataDetail.getMergedLoadName()
          .equalsIgnoreCase(segmentId)) {
        mergedSegments.add(loadMetadataDetail.getLoadName());
      }
    }
    if (!mergedSegments.isEmpty() && segmentIds.containsAll(mergedSegments)) {
      isToBeLoadedAgain = false;
    }
    return isToBeLoadedAgain;
  }

  /**
   * Provide the datamap catalog instance or null if this datamap not required to rewrite
   * the query.
   */
  public DataMapCatalog createDataMapCatalog() {
    return null;
  }

  public abstract DataMapFactory getDataMapFactory();

  public abstract boolean supportRebuild();

  public abstract boolean rebuildInternal(String newLoadName, Map<String, List<String>> segmentMap);
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.datamap;
19
20	import java.io.IOException;
21	import java.util.*;
22
23	import org.apache.carbondata.common.annotations.InterfaceAudience;
24	import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException;
25	import org.apache.carbondata.common.exceptions.sql.NoSuchDataMapException;
26	import org.apache.carbondata.common.logging.LogServiceFactory;
27	import org.apache.carbondata.core.constants.CarbonCommonConstants;
28	import org.apache.carbondata.core.datamap.dev.DataMapFactory;
29	import org.apache.carbondata.core.datamap.status.DataMapSegmentStatusUtil;
30	import org.apache.carbondata.core.datamap.status.DataMapStatusManager;
31	import org.apache.carbondata.core.locks.ICarbonLock;
32	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
33	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
34	import org.apache.carbondata.core.metadata.schema.table.DataMapSchema;
35	import org.apache.carbondata.core.metadata.schema.table.RelationIdentifier;
36	import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
37	import org.apache.carbondata.core.statusmanager.SegmentStatus;
38	import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
39	import org.apache.carbondata.core.util.path.CarbonTablePath;
40
41	import com.google.gson.Gson;
42	import org.apache.log4j.Logger;
43
44	/**
45	* DataMap is a accelerator for certain type of query. Developer can add new DataMap
46	* implementation to improve query performance.
47	*
48	* Currently two types of DataMap are supported
49	* <ol>
50	* <li> MVDataMap: materialized view type of DataMap to accelerate olap style query,
51	* like SPJG query (select, predicate, join, groupby) </li>
52	* <li> DataMap: index type of DataMap to accelerate filter query </li>
53	* </ol>
54	*
55	* <p>
56	* In following command <br>
57	* {@code CREATE DATAMAP dm ON TABLE main USING 'provider'}, <br>
58	* the <b>provider</b> string can be a short name or class name of the DataMap implementation.
59	*
60	* <br>Currently CarbonData supports following provider:
61	* <ol>
62	* <li> preaggregate: pre-aggregate table of single table </li>
63	* <li> timeseries: pre-aggregate table based on time dimension of the table </li>
64	* <li> lucene: index backed by Apache Lucene </li>
65	* <li> bloomfilter: index backed by Bloom Filter </li>
66	* </ol>
67	*
68	* @since 1.4.0
69	*/
70	@InterfaceAudience.Internal
71	public abstract class DataMapProvider {
72
73	private CarbonTable mainTable;
74	private DataMapSchema dataMapSchema;
75
76	private Logger LOGGER = LogServiceFactory.getLogService(this.getClass().getCanonicalName());	1✔
77
78	public DataMapProvider(CarbonTable mainTable, DataMapSchema dataMapSchema) {	1✔
79	this.mainTable = mainTable;	1✔
80	this.dataMapSchema = dataMapSchema;	1✔
81	}	1✔
82
83	protected final CarbonTable getMainTable() {
84	return mainTable;	1✔
85	}
86
87	public final DataMapSchema getDataMapSchema() {
88	return dataMapSchema;	1✔
89	}
90
91	/**
92	* Initialize a datamap's metadata.
93	* This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
94	* Implementation should initialize metadata for datamap, like creating table
95	*/
96	public abstract void initMeta(String ctasSqlStatement) throws MalformedDataMapCommandException,
97	IOException;
98
99	/**
100	* Initialize a datamap's data.
101	* This is called when user creates datamap, for example "CREATE DATAMAP dm ON TABLE mainTable"
102	* Implementation should initialize data for datamap, like creating data folders
103	*/
104	public void initData() { }	1✔
105
106	/**
107	* Opposite operation of {@link #initMeta(String)}.
108	* This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
109	* Implementation should clean all meta for the datamap
110	*/
111	public abstract void cleanMeta() throws IOException;
112
113	/**
114	* Opposite operation of {@link #initData()}.
115	* This is called when user drops datamap, for example "DROP DATAMAP dm ON TABLE mainTable"
116	* Implementation should clean all data for the datamap
117	*/
118	public abstract void cleanData();
119
120	/**
121	* Rebuild the datamap by loading all existing data from mainTable
122	* This is called when refreshing the datamap when
123	* 1. after datamap creation and if `autoRefreshDataMap` is set to true
124	* 2. user manually trigger REBUILD DATAMAP command
125	*/
126	public boolean rebuild() throws IOException, NoSuchDataMapException {
127	if (null == dataMapSchema.getRelationIdentifier()) {	×
128	return false;	×
129	}
130	String newLoadName = "";	×
131	String segmentMap = "";	×
132	AbsoluteTableIdentifier dataMapTableAbsoluteTableIdentifier = AbsoluteTableIdentifier	×
133	.from(dataMapSchema.getRelationIdentifier().getTablePath(),	×
134	dataMapSchema.getRelationIdentifier().getDatabaseName(),	×
135	dataMapSchema.getRelationIdentifier().getTableName());	×
136	SegmentStatusManager segmentStatusManager =	×
137	new SegmentStatusManager(dataMapTableAbsoluteTableIdentifier);
138	Map<String, List<String>> segmentMapping = new HashMap<>();	×
139	// Acquire table status lock to handle concurrent dataloading
140	ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();	×
141	try {
142	if (carbonLock.lockWithRetries()) {	×
143	LOGGER.info(	×
144	"Acquired lock for table" + dataMapSchema.getRelationIdentifier().getDatabaseName()	×
145	+ "." + dataMapSchema.getRelationIdentifier().getTableName()	×
146	+ " for table status updation");
147	String dataMapTableMetadataPath =	×
148	CarbonTablePath.getMetadataPath(dataMapSchema.getRelationIdentifier().getTablePath());	×
149	LoadMetadataDetails[] loadMetaDataDetails =	×
150	SegmentStatusManager.readLoadMetadata(dataMapTableMetadataPath);	×
151	List<LoadMetadataDetails> listOfLoadFolderDetails =	×
152	new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
153	Collections.addAll(listOfLoadFolderDetails, loadMetaDataDetails);	×
154	if (dataMapSchema.isLazy()) {	×
155	// check if rebuild to datamap is already in progress and throw exception
156	if (!listOfLoadFolderDetails.isEmpty()) {	×
157	for (LoadMetadataDetails loadMetaDetail : loadMetaDataDetails) {	×
158	if ((loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS	×
159	\|\| loadMetaDetail.getSegmentStatus()	×
160	== SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) && SegmentStatusManager	×
161	.isLoadInProgress(dataMapTableAbsoluteTableIdentifier,	×
162	loadMetaDetail.getLoadName())) {	×
163	throw new RuntimeException("Rebuild to datamap " + dataMapSchema.getDataMapName()	×
164	+ " is already in progress");
165	}
166	}
167	}
168	}
169	boolean isFullRefresh = false;	×
170	if (null != dataMapSchema.getProperties().get("full_refresh")) {	×
171	isFullRefresh = Boolean.valueOf(dataMapSchema.getProperties().get("full_refresh"));	×
172	}
173	if (!isFullRefresh) {	×
174	if (!getSpecificSegmentsTobeLoaded(segmentMapping, listOfLoadFolderDetails)) {	×
175	return false;	×
176	}
177	segmentMap = new Gson().toJson(segmentMapping);	×
178	} else {
179	List<RelationIdentifier> relationIdentifiers = dataMapSchema.getParentTables();	×
180	for (RelationIdentifier relationIdentifier : relationIdentifiers) {	×
181	List<String> mainTableSegmentList =	×
182	DataMapUtil.getMainTableValidSegmentList(relationIdentifier);	×
183	if (mainTableSegmentList.isEmpty()) {	×
184	return false;	×
185	}
186	segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT	×
187	+ relationIdentifier.getTableName(), mainTableSegmentList);	×
188	}	×
189	segmentMap = new Gson().toJson(segmentMapping);	×
190	}
191
192	// To handle concurrent dataloading to datamap, create new loadMetaEntry and
193	// set segmentMap to new loadMetaEntry and pass new segmentId with load command
194	LoadMetadataDetails loadMetadataDetail = new LoadMetadataDetails();	×
195	String segmentId =	×
196	String.valueOf(SegmentStatusManager.createNewSegmentId(loadMetaDataDetails));	×
197	loadMetadataDetail.setLoadName(segmentId);	×
198	loadMetadataDetail.setSegmentStatus(SegmentStatus.INSERT_IN_PROGRESS);	×
199	loadMetadataDetail.setExtraInfo(segmentMap);	×
200	listOfLoadFolderDetails.add(loadMetadataDetail);	×
201	newLoadName = segmentId;	×
202
203	SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath	×
204	.getTableStatusFilePath(dataMapSchema.getRelationIdentifier().getTablePath()),	×
205	listOfLoadFolderDetails
206	.toArray(new LoadMetadataDetails[listOfLoadFolderDetails.size()]));	×
207	} else {	×
208	LOGGER.error(	×
209	"Not able to acquire the lock for Table status updation for table " + dataMapSchema
210	.getRelationIdentifier().getDatabaseName() + "." + dataMapSchema	×
211	.getRelationIdentifier().getTableName());	×
212	DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName());	×
213	return false;	×
214	}
215	} finally {
216	if (carbonLock.unlock()) {	×
217	LOGGER.info("Table unlocked successfully after table status updation" + dataMapSchema	×
218	.getRelationIdentifier().getDatabaseName() + "." + dataMapSchema.getRelationIdentifier()	×
219	.getTableName());	×
220	} else {
221	LOGGER.error("Unable to unlock Table lock for table" + dataMapSchema.getRelationIdentifier()	×
222	.getDatabaseName() + "." + dataMapSchema.getRelationIdentifier().getTableName()	×
223	+ " during table status updation");
224	}
225	}	×
226	return rebuildInternal(newLoadName, segmentMapping);	×
227	}
228
229	/**
230	* This method will compare mainTable and dataMapTable segment List and loads only newly added
231	* segment from main table to dataMap table.
232	* In case if mainTable is compacted, then based on dataMap to mainTables segmentMapping, dataMap
233	* will be loaded
234	* Eg:
235	* case 1: Consider mainTableSegmentList: {0, 1, 2}, dataMapToMainTable segmentMap:
236	* { 0 -> 0, 1-> 1,2}. If (1, 2) segments of main table are compacted to 1.1 and new segment (3)
237	* is loaded to main table, then mainTableSegmentList will be updated to{0, 1.1, 3}.
238	* In this case, segment (1) of dataMap table will be marked for delete, and new segment
239	* {2 -> 1.1, 3} will be loaded to dataMap table
240	* case 2: Consider mainTableSegmentList: {0, 1, 2, 3}, dataMapToMainTable segmentMap:
241	* { 0 -> 0,1,2, 1-> 3}. If (1, 2) segments of main table are compacted to 1.1 and new segment
242	* (4) is loaded to main table, then mainTableSegmentList will be updated to {0, 1.1, 3, 4}.
243	* In this case, segment (0) of dataMap table will be marked for delete and segment (0) of
244	* main table will be added to validSegmentList which needs to be loaded again. Now, new dataMap
245	* table segment (2) with main table segmentList{2 -> 1.1, 4, 0} will be loaded to dataMap table.
246	* dataMapToMainTable segmentMap will be updated to {1 -> 3, 2 -> 1.1, 4, 0} after rebuild
247	*/
248	private boolean getSpecificSegmentsTobeLoaded(Map<String, List<String>> segmentMapping,
249	List<LoadMetadataDetails> listOfLoadFolderDetails) throws IOException {
250	List<RelationIdentifier> relationIdentifiers = dataMapSchema.getParentTables();	×
251	// invalidDataMapSegmentList holds segment list which needs to be marked for delete
252	HashSet<String> invalidDataMapSegmentList = new HashSet<>();	×
253	if (listOfLoadFolderDetails.isEmpty()) {	×
254	// If segment Map is empty, load all valid segments from main tables to dataMap
255	for (RelationIdentifier relationIdentifier : relationIdentifiers) {	×
256	List<String> mainTableSegmentList =	×
257	DataMapUtil.getMainTableValidSegmentList(relationIdentifier);	×
258	// If mainTableSegmentList is empty, no need to trigger load command
259	// TODO: handle in case of multiple tables load to datamap table
260	if (mainTableSegmentList.isEmpty()) {	×
261	return false;	×
262	}
263	segmentMapping.put(	×
264	relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT + relationIdentifier	×
265	.getTableName(), mainTableSegmentList);	×
266	}	×
267	} else {
268	for (RelationIdentifier relationIdentifier : relationIdentifiers) {	×
269	List<String> dataMapTableSegmentList = new ArrayList<>();	×
270	// Get all segments for parent relationIdentifier
271	List<String> mainTableSegmentList =	×
272	DataMapUtil.getMainTableValidSegmentList(relationIdentifier);	×
273	boolean ifTableStatusUpdateRequired = false;	×
274	for (LoadMetadataDetails loadMetaDetail : listOfLoadFolderDetails) {	×
275	if (loadMetaDetail.getSegmentStatus() == SegmentStatus.SUCCESS	×
276	\|\| loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS) {	×
277	Map<String, List<String>> segmentMaps =	×
278	DataMapSegmentStatusUtil.getSegmentMap(loadMetaDetail.getExtraInfo());	×
279	String mainTableMetaDataPath =	×
280	CarbonTablePath.getMetadataPath(relationIdentifier.getTablePath());	×
281	LoadMetadataDetails[] parentTableLoadMetaDataDetails =	×
282	SegmentStatusManager.readLoadMetadata(mainTableMetaDataPath);	×
283	String table = relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT	×
284	+ relationIdentifier.getTableName();	×
285	for (String segmentId : mainTableSegmentList) {	×
286	// In case if dataMap segment(0) is mapped to mainTable segments{0,1,2} and if
287	// {0,1,2} segments of mainTable are compacted to 0.1. Then,
288	// on next rebuild/load to dataMap, no need to load segment(0.1) again. Update the
289	// segmentMapping of dataMap segment from {0,1,2} to {0.1}
290	if (!checkIfSegmentsToBeReloaded(parentTableLoadMetaDataDetails,	×
291	segmentMaps.get(table), segmentId)) {	×
292	ifTableStatusUpdateRequired = true;	×
293	// Update loadMetaDetail with updated segment info and clear old segmentMap
294	Map<String, List<String>> updatedSegmentMap = new HashMap<>();	×
295	List<String> segmentList = new ArrayList<>();	×
296	segmentList.add(segmentId);	×
297	updatedSegmentMap.put(table, segmentList);	×
298	dataMapTableSegmentList.add(segmentId);	×
299	loadMetaDetail.setExtraInfo(new Gson().toJson(updatedSegmentMap));	×
300	segmentMaps.get(table).clear();	×
301	}
302	}	×
303	dataMapTableSegmentList.addAll(segmentMaps.get(table));	×
304	}
305	}	×
306	List<String> dataMapSegmentList = new ArrayList<>(dataMapTableSegmentList);	×
307	dataMapTableSegmentList.removeAll(mainTableSegmentList);	×
308	mainTableSegmentList.removeAll(dataMapSegmentList);	×
309	if (ifTableStatusUpdateRequired && mainTableSegmentList.isEmpty()) {	×
310	SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath	×
311	.getTableStatusFilePath(dataMapSchema.getRelationIdentifier().getTablePath()),	×
312	listOfLoadFolderDetails
313	.toArray(new LoadMetadataDetails[listOfLoadFolderDetails.size()]));	×
314	return false;	×
315	} else if (mainTableSegmentList.isEmpty()) {	×
316	return false;	×
317	}
318	if (!dataMapTableSegmentList.isEmpty()) {	×
319	List<String> invalidMainTableSegmentList = new ArrayList<>();	×
320	// validMainTableSegmentList holds segment list which needs to be loaded again
321	HashSet<String> validMainTableSegmentList = new HashSet<>();	×
322
323	// For dataMap segments which are not in main table segment list(if main table
324	// is compacted), iterate over those segments and get dataMap segments which needs to
325	// be marked for delete and main table segments which needs to be loaded again
326	for (String segmentId : dataMapTableSegmentList) {	×
327	for (LoadMetadataDetails loadMetaDetail : listOfLoadFolderDetails) {	×
328	if (loadMetaDetail.getSegmentStatus() == SegmentStatus.SUCCESS	×
329	\|\| loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS) {	×
330	Map<String, List<String>> segmentMaps =	×
331	DataMapSegmentStatusUtil.getSegmentMap(loadMetaDetail.getExtraInfo());	×
332	List<String> segmentIds = segmentMaps.get(	×
333	relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT	×
334	+ relationIdentifier.getTableName());	×
335	if (segmentIds.contains(segmentId)) {	×
336	segmentIds.remove(segmentId);	×
337	validMainTableSegmentList.addAll(segmentIds);	×
338	invalidMainTableSegmentList.add(segmentId);	×
339	invalidDataMapSegmentList.add(loadMetaDetail.getLoadName());	×
340	}
341	}
342	}	×
343	}	×
344	// remove invalid segment from validMainTableSegmentList if present
345	validMainTableSegmentList.removeAll(invalidMainTableSegmentList);	×
346	// Add all valid segments of main table which needs to be loaded again
347	mainTableSegmentList.addAll(validMainTableSegmentList);	×
348	segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT	×
349	+ relationIdentifier.getTableName(), mainTableSegmentList);	×
350	} else {	×
351	segmentMapping.put(relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT	×
352	+ relationIdentifier.getTableName(), mainTableSegmentList);	×
353	}
354	}	×
355	}
356	// Remove invalid datamap segments
357	if (!invalidDataMapSegmentList.isEmpty()) {	×
358	for (LoadMetadataDetails loadMetadataDetail : listOfLoadFolderDetails) {	×
359	if (invalidDataMapSegmentList.contains(loadMetadataDetail.getLoadName())) {	×
360	loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);	×
361	}
362	}	×
363	}
364	return true;	×
365	}
366
367	/**
368	* This method checks if dataMap table segment has to be reloaded again or not
369	*/
370	private boolean checkIfSegmentsToBeReloaded(LoadMetadataDetails[] loadMetaDataDetails,
371	List<String> segmentIds, String segmentId) {
372	boolean isToBeLoadedAgain = true;	×
373	List<String> mergedSegments = new ArrayList<>();	×
374	for (LoadMetadataDetails loadMetadataDetail : loadMetaDataDetails) {	×
375	if (null != loadMetadataDetail.getMergedLoadName() && loadMetadataDetail.getMergedLoadName()	×
376	.equalsIgnoreCase(segmentId)) {	×
377	mergedSegments.add(loadMetadataDetail.getLoadName());	×
378	}
379	}
380	if (!mergedSegments.isEmpty() && segmentIds.containsAll(mergedSegments)) {	×
381	isToBeLoadedAgain = false;	×
382	}
383	return isToBeLoadedAgain;	×
384	}
385
386	/**
387	* Provide the datamap catalog instance or null if this datamap not required to rewrite
388	* the query.
389	*/
390	public DataMapCatalog createDataMapCatalog() {
391	return null;	×
392	}
393
394	public abstract DataMapFactory getDataMapFactory();
395
396	public abstract boolean supportRebuild();
397
398	public abstract boolean rebuildInternal(String newLoadName, Map<String, List<String>> segmentMap);
399	}

apache / carbondata / 11970

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous