3514

Build # 3514

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/428c2c47e512ba19626242d6c851fdaf47c2cb93">428c2c47e<a href="https://github.com/apache/carbondata/commit/428c2c47e512ba19626242d6c851fdaf47c2cb93">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/e50993f62cc5a67f43f38870e33556923f272747">e50993f62</a>">e50993f62</a><a href="https://github.com/apache/carbondata/commit/428c2c47e512ba19626242d6c851fdaf47c2cb93"> into df71291ec">df71291ec</a>

Run Details

54 of 54 new or added lines in 5 files covered. (100.0%)

64130 of 81965 relevant lines covered (78.24%)

1.04 hits per line

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

5.63

/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.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());
      }
    } 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<>();
        for (LoadMetadataDetails loadMetaDetail : listOfLoadFolderDetails) {
          if (loadMetaDetail.getSegmentStatus() == SegmentStatus.SUCCESS
              || loadMetaDetail.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS) {
            Map<String, List<String>> segmentMaps =
                DataMapSegmentStatusUtil.getSegmentMap(loadMetaDetail.getExtraInfo());
            dataMapTableSegmentList.addAll(segmentMaps.get(
                relationIdentifier.getDatabaseName() + CarbonCommonConstants.POINT
                    + relationIdentifier.getTableName()));
          }
        }
        List<String> dataMapSegmentList = new ArrayList<>(dataMapTableSegmentList);
        // Get all segments for parent relationIdentifier
        List<String> mainTableSegmentList =
            DataMapUtil.getMainTableValidSegmentList(relationIdentifier);
        dataMapTableSegmentList.removeAll(mainTableSegmentList);
        mainTableSegmentList.removeAll(dataMapSegmentList);
        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;
  }

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

apache / carbondata / 3514

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